8680627292

Committed 14 Apr 2024 05:23PM UTC coverage: 98.814% (+0.003%) from 98.811%

Build Type

github

Committed by Giorgio Salluzzo

Commit Message

Add getblocking() method.

Pull Request Pull Request #228: Add getblocking() method

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

833 of 843 relevant lines covered (98.81%)

5.88 hits per line

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

100.0

/mocket/plugins/httpretty/__init__.py

from mocket import Mocket, mocketize
from mocket.async_mocket import async_mocketize
from mocket.compat import ENCODING, byte_type, text_type
from mocket.mockhttp import Entry as MocketHttpEntry
from mocket.mockhttp import Request as MocketHttpRequest
from mocket.mockhttp import Response as MocketHttpResponse


def httprettifier_headers(headers):
    return {k.lower().replace("_", "-"): v for k, v in headers.items()}


class Request(MocketHttpRequest):
    @property
    def body(self):
        return super().body.encode(ENCODING)

    @property
    def headers(self):
        return httprettifier_headers(super().headers)


class Response(MocketHttpResponse):
    def get_protocol_data(self, str_format_fun_name="lower"):
        if "server" in self.headers and self.headers["server"] == "Python/Mocket":
            self.headers["server"] = "Python/HTTPretty"
        return super(Response, self).get_protocol_data(
            str_format_fun_name=str_format_fun_name
        )

    def set_base_headers(self):
        super(Response, self).set_base_headers()
        self.headers = httprettifier_headers(self.headers)

    original_set_base_headers = set_base_headers

    def set_extra_headers(self, headers):
        self.headers.update(headers)


class Entry(MocketHttpEntry):
    request_cls = Request
    response_cls = Response


activate = mocketize
httprettified = mocketize
async_httprettified = async_mocketize

enable = Mocket.enable
disable = Mocket.disable
reset = Mocket.reset

GET = Entry.GET
PUT = Entry.PUT
POST = Entry.POST
DELETE = Entry.DELETE
HEAD = Entry.HEAD
PATCH = Entry.PATCH
OPTIONS = Entry.OPTIONS


def register_uri(
    method,
    uri,
    body="HTTPretty :)",
    adding_headers=None,
    forcing_headers=None,
    status=200,
    responses=None,
    match_querystring=False,
    priority=0,
    **headers,
):
    headers = httprettifier_headers(headers)

    if adding_headers is not None:
        headers.update(httprettifier_headers(adding_headers))

    if forcing_headers is not None:

        def force_headers(self):
            self.headers = httprettifier_headers(forcing_headers)

        Response.set_base_headers = force_headers
    else:
        Response.set_base_headers = Response.original_set_base_headers

    if responses:
        Entry.register(method, uri, *responses)
    else:
        Entry.single_register(
            method,
            uri,
            body=body,
            status=status,
            headers=headers,
            match_querystring=match_querystring,
        )


class MocketHTTPretty:
    Response = Response

    def __getattr__(self, name):
        if name == "last_request":
            return Mocket.last_request()
        if name == "latest_requests":
            return Mocket.request_list()
        return getattr(Entry, name)


HTTPretty = MocketHTTPretty()
HTTPretty.register_uri = register_uri
httpretty = HTTPretty

__all__ = (
    "HTTPretty",
    "httpretty",
    "activate",
    "async_httprettified",
    "httprettified",
    "enable",
    "disable",
    "reset",
    "Response",
    "GET",
    "PUT",
    "POST",
    "DELETE",
    "HEAD",
    "PATCH",
    "register_uri",
    "text_type",
    "byte_type",
)

1	from mocket import Mocket, mocketize	6✔
2	from mocket.async_mocket import async_mocketize	6✔
3	from mocket.compat import ENCODING, byte_type, text_type	6✔
4	from mocket.mockhttp import Entry as MocketHttpEntry	6✔
5	from mocket.mockhttp import Request as MocketHttpRequest	6✔
6	from mocket.mockhttp import Response as MocketHttpResponse	6✔
7
8
9	def httprettifier_headers(headers):	6✔
10	return {k.lower().replace("_", "-"): v for k, v in headers.items()}	6✔
11
12
13	class Request(MocketHttpRequest):	6✔
14	@property	6✔
15	def body(self):	6✔
16	return super().body.encode(ENCODING)	6✔
17
18	@property	6✔
19	def headers(self):	6✔
20	return httprettifier_headers(super().headers)	6✔
21
22
23	class Response(MocketHttpResponse):	6✔
24	def get_protocol_data(self, str_format_fun_name="lower"):	6✔
25	if "server" in self.headers and self.headers["server"] == "Python/Mocket":	6✔
26	self.headers["server"] = "Python/HTTPretty"	6✔
27	return super(Response, self).get_protocol_data(	6✔
28	str_format_fun_name=str_format_fun_name
29	)
30
31	def set_base_headers(self):	6✔
32	super(Response, self).set_base_headers()	6✔
33	self.headers = httprettifier_headers(self.headers)	6✔
34
35	original_set_base_headers = set_base_headers	6✔
36
37	def set_extra_headers(self, headers):	6✔
38	self.headers.update(headers)	6✔
39
40
41	class Entry(MocketHttpEntry):	6✔
42	request_cls = Request	6✔
43	response_cls = Response	6✔
44
45
46	activate = mocketize	6✔
47	httprettified = mocketize	6✔
48	async_httprettified = async_mocketize	6✔
49
50	enable = Mocket.enable	6✔
51	disable = Mocket.disable	6✔
52	reset = Mocket.reset	6✔
53
54	GET = Entry.GET	6✔
55	PUT = Entry.PUT	6✔
56	POST = Entry.POST	6✔
57	DELETE = Entry.DELETE	6✔
58	HEAD = Entry.HEAD	6✔
59	PATCH = Entry.PATCH	6✔
60	OPTIONS = Entry.OPTIONS	6✔
61
62
63	def register_uri(	6✔
64	method,
65	uri,
66	body="HTTPretty :)",
67	adding_headers=None,
68	forcing_headers=None,
69	status=200,
70	responses=None,
71	match_querystring=False,
72	priority=0,
73	**headers,
74	):
75	headers = httprettifier_headers(headers)	6✔
76
77	if adding_headers is not None:	6✔
78	headers.update(httprettifier_headers(adding_headers))	6✔
79
80	if forcing_headers is not None:	6✔
81
82	def force_headers(self):	6✔
83	self.headers = httprettifier_headers(forcing_headers)	6✔
84
85	Response.set_base_headers = force_headers	6✔
86	else:
87	Response.set_base_headers = Response.original_set_base_headers	6✔
88
89	if responses:	6✔
90	Entry.register(method, uri, *responses)	6✔
91	else:
92	Entry.single_register(	6✔
93	method,
94	uri,
95	body=body,
96	status=status,
97	headers=headers,
98	match_querystring=match_querystring,
99	)
100
101
102	class MocketHTTPretty:	6✔
103	Response = Response	6✔
104
105	def __getattr__(self, name):	6✔
106	if name == "last_request":	6✔
107	return Mocket.last_request()	6✔
108	if name == "latest_requests":	6✔
109	return Mocket.request_list()	6✔
110	return getattr(Entry, name)	6✔
111
112
113	HTTPretty = MocketHTTPretty()	6✔
114	HTTPretty.register_uri = register_uri	6✔
115	httpretty = HTTPretty	6✔
116
117	__all__ = (	6✔
118	"HTTPretty",
119	"httpretty",
120	"activate",
121	"async_httprettified",
122	"httprettified",
123	"enable",
124	"disable",
125	"reset",
126	"Response",
127	"GET",
128	"PUT",
129	"POST",
130	"DELETE",
131	"HEAD",
132	"PATCH",
133	"register_uri",
134	"text_type",
135	"byte_type",
136	)

mindflayer / python-mocket / 8680627292

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