python开源项目源码解读---httmock，专门为requests 设计的mock 工具

1. python开源项目httmock

httmock 是一款专门为requests 设计的mock工具，在测试阶段，它可以很好的为接口mock数据，降低准备测试数据的难度，git 地址：https://github.com/patrys/httmock ，下面其hub主页上的简单示例

from httmock import urlmatch, HTTMock
import requests

@urlmatch(netloc=r'(.*\.)?google\.com$')
def google_mock(url, request):
    return 'Feeling lucky, punk?'

with HTTMock(google_mock):
    r = requests.get('http://google.com/')

print(r.content)

请求的url是http://google.com/， 但得道的却是google_mock函数的返回值，我并不关心这种mock数据在实际开发中的应用价值，我比较好奇的是它是如何做到mock数据的。

很显然，HTTMock 一定是在requests发送http请求的过程中走了某种hook的操作，请求并没有真实的发送给目标url，而是转向了google_mock函数，那么它是如何做到的呢？

2. HTTMock 上下文管理器

with HTTMock(google_mock):
    r = requests.get('http://google.com/')

HTTMock 是上下文管理器，它实现了__enter__() 和__exit__() ，只有在with语句中，requests发出的请求才能被hook，改变请求目标，现在，进入HTTMock 内部一探究竟

class HTTMock(object):
    """
    Acts as a context manager to allow mocking
    """
    STATUS_CODE = 200

    def __init__(self, *handlers):
        self.handlers = handlers

    def __enter__(self):
        self._real_session_send = requests.Session.send
        self._real_session_prepare_request = requests.Session.prepare_request

        for handler in self.handlers:
            handler_clean_call(handler)

        def _fake_send(session, request, **kwargs):
            response = self.intercept(request, **kwargs)

            if isinstance(response, requests.Response):
                # this is pasted from requests to handle redirects properly:
                kwargs.setdefault('stream', session.stream)
                kwargs.setdefault('verify', session.verify)
                kwargs.setdefault('cert', session.cert)
                kwargs.setdefault('proxies', session.proxies)

                allow_redirects = kwargs.pop('allow_redirects', True)
                stream = kwargs.get('stream')
                timeout = kwargs.get('timeout')
                verify = kwargs.get('verify')
                cert = kwargs.get('cert')
                proxies = kwargs.get('proxies')

                gen = session.resolve_redirects(
                    response,
                    request,
                    stream=stream,
                    timeout=timeout,
                    verify=verify,
                    cert=cert,
                    proxies=proxies)

                history = [resp for resp in gen] if allow_redirects else []

                if history:
                    history.insert(0, response)
                    response = history.pop()
                    response.history = tuple(history)

                session.cookies.update(response.cookies)

                return response

            return self._real_session_send(session, request, **kwargs)

        def _fake_prepare_request(session, request):
            """
            Fake this method so the `PreparedRequest` objects contains
            an attribute `original` of the original request.
            """
            prep = self._real_session_prepare_request(session, request)
            prep.original = request
            return prep

        requests.Session.send = _fake_send
        requests.Session.prepare_request = _fake_prepare_request

        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        requests.Session.send = self._real_session_send
        requests.Session.prepare_request = self._real_session_prepare_request

    def intercept(self, request, **kwargs):
        url = urlparse.urlsplit(request.url)
        res = first_of(self.handlers, url, request)

        if isinstance(res, requests.Response):
            return res
        elif isinstance(res, dict):
            return response(res.get('status_code'),
                            res.get('content'),
                            res.get('headers'),
                            res.get('reason'),
                            res.get('elapsed', 0),
                            request,
                            stream=kwargs.get('stream', False),
                            http_vsn=res.get('http_vsn', 11))
        elif isinstance(res, (text_type, binary_type)):
            return response(content=res, stream=kwargs.get('stream', False))
        elif res is None:
            return None
        else:
            raise TypeError(
                "Dont know how to handle response of type {0}".format(type(res)))

鉴于代码量并不是很大，我这里全部贴出来。

2.1 enter()

进入上下文环境时，执行__enter__() ， 在这里，做了两个特别关键的操作

self._real_session_send = requests.Session.send
self._real_session_prepare_request = requests.Session.prepare_request

......

requests.Session.send = _fake_send
requests.Session.prepare_request = _fake_prepare_request

requests 发送请求之前，会使用prepare_request 函数生成reqeust对象，然后使用send方法发送请求，在行__enter__() 中，作者保存好原始的send 和 prepare_request 方法，然后将requests.Session.send 替换为_fake_send， requests.Session.prepare_request 替换为 _fake_prepare_request。

这样一来，就等于修改了requests的源码，不过这种修改只在当前进程中生效，那么随后的prepare_request 和 send的过程，就都是使用作者提供的方法，_fake_send 和 _fake_prepare_request 。

在 _fake_send 方法中，调用了intercept方法

    def intercept(self, request, **kwargs):
        url = urlparse.urlsplit(request.url)
        res = first_of(self.handlers, url, request)

def first_of(handlers, *args, **kwargs):
    for handler in handlers:
        res = handler(*args, **kwargs)
        if res is not None:
            return res

self.handlers 正是创建HTTMock上下文时传入的google_mock参数，返回结果正是google_mock 被调用执行后的返回值，这样就完成了对requests库的一次hook操作。

2.2 exit

ef __exit__(self, exc_type, exc_val, exc_tb):
        requests.Session.send = self._real_session_send
        requests.Session.prepare_request = self._real_session_prepare_request

在退出上下文时，需要将requests.Session.send 和 requests.Session.prepare_request 修改为原始的方法，这样其他mock请求能够正常调用，由此可见，这个mock工具不能够在多线程中使用。