Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_same_url_different_methods(self):
url = "http://bit.ly/fakeurl"
response_to_mock = {"content": 0, "method": None}
responses = []
methods = [Entry.PUT, Entry.GET, Entry.POST]
for m in methods:
response_to_mock["method"] = m
Entry.single_register(m, url, body=json.dumps(response_to_mock))
response_to_mock["content"] += 1
for m in methods:
responses.append(requests.request(m, url).json())
methods_from_responses = [r["method"] for r in responses]
contents_from_responses = [r["content"] for r in responses]
self.assertEqual(methods, methods_from_responses)
self.assertEqual(list(range(len(methods))), contents_from_responses)
def test_http_session(self):
url = 'http://httpbin.org/ip'
body = "asd" * 100
Entry.single_register(Entry.GET, url, body=body, status=404)
Entry.single_register(Entry.POST, url, body=body*2, status=201)
async def main(l):
async with aiohttp.ClientSession(loop=l) as session:
with async_timeout.timeout(3):
async with session.get(url) as get_response:
assert get_response.status == 404
assert await get_response.text() == body
with async_timeout.timeout(3):
async with session.post(url, data=body * 6) as post_response:
assert post_response.status == 201
assert await post_response.text() == body * 2
assert Mocket.last_request().method == 'POST'
assert Mocket.last_request().body == body * 6
def test_register(self):
with mock.patch("time.gmtime") as tt:
tt.return_value = time.struct_time((2013, 4, 30, 10, 39, 21, 1, 120, 0))
Entry.register(
Entry.GET,
"http://testme.org/get?a=1&b=2#test",
Response('{"a": "€"}', headers={"content-type": "application/json"}),
)
entries = Mocket._entries[("testme.org", 80)]
self.assertEqual(len(entries), 1)
entry = entries[0]
self.assertEqual(entry.method, "GET")
self.assertEqual(entry.schema, "http")
self.assertEqual(entry.path, "/get")
self.assertEqual(entry.query, "a=1&b=2")
self.assertEqual(len(entry.responses), 1)
response = entry.responses[0]
self.assertEqual(response.body, b'{"a": "\xe2\x82\xac"}')
self.assertEqual(response.status, 200)
self.assertEqual(
def test_sendall(self):
with mock.patch("time.gmtime") as tt:
tt.return_value = time.struct_time((2013, 4, 30, 10, 39, 21, 1, 120, 0))
Entry.single_register(
Entry.GET, "http://testme.org/get/p/?a=1&b=2", body="test_body"
)
resp = urlopen("http://testme.org/get/p/?b=2&a=1", timeout=10)
self.assertEqual(resp.code, 200)
self.assertEqual(resp.read(), b"test_body")
self.assertEqualHeaders(
dict(resp.headers),
{
"Status": "200",
"Content-length": "9",
"Server": "Python/Mocket",
"Connection": "close",
"Date": "Tue, 30 Apr 2013 10:39:21 GMT",
"Content-type": "text/plain; charset=utf-8",
},
)
self.assertEqual(len(Mocket._requests), 1)
def _parse_requestline(line):
"""
http://www.w3.org/Protocols/rfc2616/rfc2616-sec5.html#sec5
>>> Entry._parse_requestline('GET / HTTP/1.0') == ('GET', '/', '1.0')
True
>>> Entry._parse_requestline('post /testurl htTP/1.1') == ('POST', '/testurl', '1.1')
True
>>> Entry._parse_requestline('Im not a RequestLine')
Traceback (most recent call last):
...
ValueError: Not a Request-Line
"""
m = re.match(r'({})\s+(.*)\s+HTTP/(1.[0|1])'.format('|'.join(Entry.METHODS)), line, re.I)
if m:
return m.group(1).upper(), m.group(2), m.group(3)
else:
raise ValueError('Not a Request-Line')
header_lines = CRLF.join(
["{0}: {1}".format(k.lower(), v) for k, v in self.headers.items()]
)
return "{0}\r\n{1}\r\n\r\n".format(status_line, header_lines).encode("utf-8")
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):
response_cls = Response
activate = mocketize
httprettified = 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
from pook.engine import MockEngine
from pook.interceptors.base import BaseInterceptor
from mocket.mocket import Mocket
from mocket.mockhttp import Entry, Response
class MocketPookEntry(Entry):
pook_request = None
pook_engine = None
def can_handle(self, data):
can_handle = super(MocketPookEntry, self).can_handle(data)
if can_handle:
self.pook_engine.match(self.pook_request)
return can_handle
@classmethod
def single_register(cls, method, uri, body='', status=200, headers=None, match_querystring=True):
entry = cls(
uri, method, Response(
body=body, status=status, headers=headers
), match_querystring=match_querystring