How to use the mocket.mockhttp.Entry function in mocket

To help you get started, we’ve selected a few mocket examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github mindflayer / python-mocket / tests / main / test_http.py View on Github external
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)
github mindflayer / python-mocket / tests / tests35 / test_http_aiohttp.py View on Github external
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
github mindflayer / python-mocket / tests / main / test_http.py View on Github external
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(
github mindflayer / python-mocket / tests / main / test_http.py View on Github external
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)
github mindflayer / python-mocket / mocket / mockhttp.py View on Github external
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')
github mindflayer / python-mocket / mocket / plugins / httpretty / __init__.py View on Github external
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
github mindflayer / python-mocket / mocket / plugins / pook_mock_engine.py View on Github external
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