Examples

Basic mocking example using requests

import pook
import requests


# Enable mock engine
pook.on()

pook.get(
    "httpbin.org/ip",
    reply=403,
    response_type="json",
    response_headers={"server": "pook"},
    response_json={"error": "not found"},
)

pook.get(
    "httpbin.org/headers",
    reply=404,
    response_type="json",
    response_headers={"server": "pook"},
    response_json={"error": "not found"},
)

res = requests.get("http://httpbin.org/ip")
print("Status:", res.status_code)
print("Headers:", res.headers)
print("Body:", res.json())

res = requests.get("http://httpbin.org/headers")
print("Status:", res.status_code)
print("Headers:", res.headers)
print("Body:", res.json())

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())
print("Unmatched requests:", pook.unmatched_requests())

Chainable API DSL

import json
import pook
import requests


# Enable mock engine
pook.on()

(
    pook.post("httpbin.org/post")
    .json({"foo": "bar"})
    .type("json")
    .header("Client", "requests")
    .reply(204)
    .headers({"server": "pook"})
    .json({"error": "simulated"})
)

res = requests.post(
    "http://httpbin.org/post",
    data=json.dumps({"foo": "bar"}),
    headers={"Client": "requests", "Content-Type": "application/json"},
)

print("Status:", res.status_code)
print("Body:", res.json())

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

Context manager for isolated mocking

import pook
import requests


with pook.api.context():
    pook.get(
        "httpbin.org/ip",
        reply=403,
        response_headers={"pepe": "lopez"},
        response_json={"error": "not found"},
    )

    res = requests.get("http://httpbin.org/ip")
    print("Status:", res.status_code)
    print("Headers:", res.headers)
    print("Body:", res.json())

    print("Is done:", pook.isdone())
    print("Pending mocks:", pook.pending_mocks())
    print("Unmatched requests:", pook.unmatched_requests())

Single mock context manager definition for isolated mocking

import pook
import requests


# Define a new mock that will be only active within the context manager
with pook.get(
    "httpbin.org/ip",
    reply=403,
    response_headers={"pepe": "lopez"},
    response_json={"error": "not found"},
) as mock:
    res = requests.get("http://httpbin.org/ip")
    print("#1 Status:", res.status_code)
    print("#1 Headers:", res.headers)
    print("#1 Body:", res.json())
    print("----------------")

    res = requests.get("http://httpbin.org/ip")
    print("#2 Status:", res.status_code)
    print("#2 Headers:", res.headers)
    print("#2 Body:", res.json())
    print("----------------")

    print("Mock is done:", mock.isdone())
    print("Mock matches:", mock.total_matches)

    print("Is done:", pook.isdone())
    print("Pending mocks:", pook.pending_mocks())
    print("Unmatched requests:", pook.unmatched_requests())


# Explicitly disable mock engine
pook.off()

# Perform a real HTTP request since we are running the
# request outside of the context manager
res = requests.get("http://httpbin.org/ip")
print("#3 Status:", res.status_code)
print("#3 Headers:", res.headers)
print("#3 Body:", res.json())

Declaring mocks as decorators

import pook
import requests


@pook.get("http://httpbin.org/status/500", reply=204)
@pook.get("http://httpbin.org/status/400", reply=200, persist=True)
def fetch(url):
    return requests.get(url)


# Test function
res = fetch("http://httpbin.org/status/400")
print("#1 status:", res.status_code)

res = fetch("http://httpbin.org/status/500")
print("#2 status:", res.status_code)

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

# Disable mock engine
pook.off()

# Test real request
res = requests.get("http://httpbin.org/status/400")
print("Test status:", res.status_code)

Activating the mock engine via decorator within the function context

import pook
import requests


@pook.on
def run():
    pook.get(
        "httpbin.org/ip",
        reply=403,
        response_headers={"pepe": "lopez"},
        response_json={"error": "not found"},
    )

    res = requests.get("http://httpbin.org/ip")
    print("Status:", res.status_code)
    print("Headers:", res.headers)
    print("Body:", res.json())

    print("Is done:", pook.isdone())
    print("Pending mocks:", pook.pending_mocks())
    print("Unmatched requests:", pook.unmatched_requests())


run()

Activating the mock engine via decorator within an async coroutine function context

# flake8: noqa
import pook
import asyncio
import aiohttp


@pook.on
async def run():
    pook.get(
        "httpbin.org/ip",
        reply=403,
        response_headers={"pepe": "lopez"},
        response_json={"error": "not found"},
    )

    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.get("http://httpbin.org/ip") as res:
            print("Status:", res.status)
            print("Headers:", res.headers)
            print("Body:", await res.text())

            print("Is done:", pook.isdone())
            print("Pending mocks:", pook.pending_mocks())
            print("Unmatched requests:", pook.unmatched_requests())


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Featured JSON body matching

import json
import pook
import requests

# Enable mock engine
pook.on()

(
    pook.post("httpbin.org/post")
    .json({"foo": "bar"})
    .reply(204)
    .json({"error": "simulated"})
)

res = requests.post("http://httpbin.org/post", data=json.dumps({"foo": "bar"}))
print("Status:", res.status_code)
print("Body:", res.json())

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

JSONSchema based body matching

import json
import pook
import requests


schema = {
    "type": "object",
    "properties": {
        "foo": {"type": "string"},
    },
}

# Enable mock engine
pook.on()

(
    pook.post("httpbin.org/post")
    .jsonschema(schema)
    .reply(204)
    .json({"error": "simulated"})
)

res = requests.post("http://httpbin.org/post", data=json.dumps({"foo": "bar"}))

print("Status:", res.status_code)
print("Body:", res.json())

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

Request Query Params matching

import pook
import requests

# Enable mock engine
pook.on()

(
    pook.get("httpbin.org/get")
    .params({"foo": "bar"})
    .reply(204)
    .json({"error": "simulated"})
)

res = requests.get("http://httpbin.org/get", params={"foo": "bar"})

assert res.status_code == 204
assert res.json() == {"error": "simulated"}
assert pook.isdone()
assert pook.pending_mocks() == []

Enable real networking mode

import pook
import requests


# Enable mock engine
pook.on()

# Enable network mode
pook.enable_network()

(
    pook.get("httpbin.org/headers")
    .reply(204)
    .headers({"server": "pook"})
    .json({"error": "simulated"})
)

res = requests.get("http://httpbin.org/headers")
print("Mock status:", res.status_code)

# Real network request, since pook cannot match any mock
res = requests.get("http://httpbin.org/ip")
print("Real status:", res.status_code)

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

# Disable network mode once we're done
pook.disable_network()

Persistent mock

import pook
import requests


# Enable mock engine
pook.on()

(
    pook.get("httpbin.org")
    .headers({"Client": "requests"})
    .persist()
    .reply(400)
    .headers({"server": "pook"})
    .json({"error": "simulated"})
)

res = requests.get("http://httpbin.org", headers={"Client": "requests"})

print("Status:", res.status_code)
print("Headers:", res.headers)
print("Body:", res.json())

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

Time TTL limited mock

import pook
import requests


# Enable mock engine
pook.on()

# Declare mock
(
    pook.get("httpbin.org")
    .times(2)
    .reply(400)
    .headers({"server": "pook"})
    .json({"error": "simulated"})
)

# Mock request 1
res = requests.get("http://httpbin.org")
print("#1 status:", res.status_code)
print("#1 body:", res.json())

# Mock request 2
res = requests.get("http://httpbin.org")
print("#2 status:", res.status_code)
print("#2 body:", res.json())

# Real request 3
try:
    requests.get("http://httpbin.org")
except Exception:
    print("Request #3 not matched due to expired mock")

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

Regular expression matching

import pook
import requests


# Enable mock engine
pook.on()

# Mock definition based
(
    pook.get(pook.regex("h[t]{2}pbin.*"))
    .path(pook.regex("/foo/[a-z]+/baz"))
    .header("Client", pook.regex("requests|pook"))
    .times(2)
    .reply(200)
    .headers({"server": "pook"})
    .json({"foo": "bar"})
)

# Perform request
res = requests.get("http://httpbin.org/foo/bar/baz", headers={"Client": "requests"})
print("Status:", res.status_code)
print("Body:", res.json())

# Perform second request
res = requests.get("http://httpbin.org/foo/foo/baz", headers={"Client": "pook"})
print("Status:", res.status_code)
print("Body:", res.json())

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())

unittest integration

import pook
import unittest
import requests


class TestUnitTestEngine(unittest.TestCase):
    @pook.on
    def test_request(self):
        pook.get("server.com/foo").reply(204)
        res = requests.get("http://server.com/foo")
        self.assertEqual(res.status_code, 204)

    def test_request_with_context_manager(self):
        with pook.use():
            pook.get("server.com/bar", reply=204)
            res = requests.get("http://server.com/bar")
            self.assertEqual(res.status_code, 204)

    @pook.on
    def test_no_match_exception(self):
        pook.get("server.com/bar", reply=204)
        try:
            requests.get("http://server.com/baz")
        except Exception:
            pass
        else:
            raise RuntimeError("expected to fail")

py.test integration

import pook
import pytest
import requests


@pook.activate
def test_simple_pook_request():
    pook.get("server.com/foo").reply(204)
    res = requests.get("http://server.com/foo")
    assert res.status_code == 204


@pook.on
def test_enable_engine():
    pook.get("server.com/foo").reply(204)
    res = requests.get("http://server.com/foo")
    assert res.status_code == 204
    pook.disable()


@pook.get("server.com/bar", reply=204)
def test_decorator():
    res = requests.get("http://server.com/bar")
    assert res.status_code == 204


def test_context_manager():
    with pook.use():
        pook.get("server.com/baz", reply=204)
        res = requests.get("http://server.com/baz")
        assert res.status_code == 204


@pook.on
def test_no_match_exception():
    pook.get("server.com/bar", reply=204)
    with pytest.raises(Exception):
        requests.get("http://server.com/baz")

Simulated error exception on mock matching

import pook
import requests


# Enable mock engine
pook.on()

# Simulated error exception on request matching
pook.get("httpbin.org/status/500", error=Exception("simulated error"))

try:
    requests.get("http://httpbin.org/status/500")
except Exception as err:
    print("Error:", err)

Using urllib3 as HTTP client

import pook
import urllib3


# Mock HTTP traffic only in the given context
with pook.use():
    pook.get("http://httpbin.org/status/404").reply(204)

    # Intercept request
    http = urllib3.PoolManager()
    r = http.request("GET", "http://httpbin.org/status/404")
    print("#1 status:", r.status)


# Real request outside of the context manager
http = urllib3.PoolManager()
r = http.request("GET", "http://httpbin.org/status/404")
print("#2 status:", r.status)

Using urllib3 to return a chunked response

import pook
import urllib3


# Mock HTTP traffic only in the given context
with pook.use():
    (
        pook.get("httpbin.org/chunky")
        .reply(200)
        .body(["returned", "as", "chunks"], chunked=True)
    )

    # Intercept request
    http = urllib3.PoolManager()
    r = http.request("GET", "httpbin.org/chunky")
    print("Chunks:", list(r.read_chunked()))

Asynchronous HTTP request using aiohttp

# flake8: noqa

import pook
import aiohttp
import asyncio
import async_timeout


async def fetch(session, url, data):
    with async_timeout.timeout(10):
        async with session.get(url, data=data) as res:
            print("Status:", res.status)
            print("Headers:", res.headers)
            print("Body:", await res.text())


with pook.use(network=True):
    pook.get(
        "http://httpbin.org/ip",
        reply=404,
        response_type="json",
        response_headers={"Server": "nginx"},
        response_json={"error": "not found"},
    )

    async def main(loop):
        async with aiohttp.ClientSession(loop=loop) as session:
            await fetch(session, "http://httpbin.org/ip", bytearray("foo bar", "utf-8"))

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))

Using http.client standard Python package as HTTP client

import http.client
import pook


# Enable mock engine
pook.on()

mock = pook.get(
    "http://httpbin.org/ip",
    reply=404,
    response_type="json",
    response_json={"error": "not found"},
)

conn = http.client.HTTPConnection("httpbin.org")
conn.request("GET", "/ip")

res = conn.getresponse()
print("Status:", res.status, res.reason)
print("Headers:", res.headers)
print("Body:", res.read())
print("Mock calls:", mock.calls)

print("Is done:", pook.isdone())
print("Pending mocks:", pook.pending_mocks())
print("Unmatched requests:", pook.unmatched_requests())

Example using mocket Python library as underlying mock engine

# Be sure you have `mocket` installed:
# $ pip install mocket

import pook
import requests
from mocket.plugins.pook_mock_engine import MocketEngine

# Use mocket library as underlying mock engine
pook.set_mock_engine(MocketEngine)

# Explicitly enable pook HTTP mocking (required)
pook.on()

# Target server URL to mock out
url = "http://twitter.com/api/1/foobar"

# Define your mock
mock = pook.get(
    url,
    reply=404,
    times=2,
    headers={"content-type": "application/json"},
    response_json={"error": "foo"},
)

# Run first HTTP request
requests.get(url)
assert mock.calls == 1

# Run second HTTP request
res = requests.get(url)
assert mock.calls == 2

# Assert response data
assert res.status_code == 404
assert res.json() == {"error": "foo"}

# Explicitly disable pook (optional)
pook.off()

Hy programming language example

(import [pook])
(import [requests])

(defn request [url &optional [status 404]]
  (doto (.mock pook url) (.reply status))
  (let [res (.get requests url)]
    (. res status_code)))

(defn run []
  (with [(.use pook)]
    (print "Status:" (request "http://foo.com/bar" :status 204))))

;; Run test program
(defmain [&args] (run))