Python responses 模块，activate() 实例源码

def test_handle_raises_error_if_status_is_400(api_call):
    responses.add(api_call.method, api_call.url, body=HTTPError())
    with raises(HTTPError):
        handle(api_call())


# @responses.activate
# @mark.parametrize('api_call', api_calls)
# def test_call_endpoint_with_query_parameters(api_call):
#     responses.add(
#         api_call.method,
#         api_call.url + '?spam=eggs&foo=bar',
#         match_querystring=False
#     )
#     params = dict(foo='bar', spam='eggs')
#     assert api_call(params=params).status_code == 200

def feed_response(sources, modifier=None):
    def decorator(func):
        @wraps(func)
        @responses.activate
        def func_wrapper(obj, *args, **kwargs):
            source_list = sources
            feeds = []
            if type(sources) is not list:
                source_list = [sources]

            for source in source_list:
                if source == 'twitter':
                    feeds.append(_twitter(modifier))
                elif source == 'instagram':
                    feeds.append(_instagram(modifier))
                elif source == 'facebook':
                    feeds.append(_facebook(modifier))
            feeds.extend(args)
            return func(obj, *feeds, **kwargs)
        return func_wrapper
    return decorator

def mock(func):
        @responses.activate
        def wrapped(self, *args, **kwargs):
            self._setup_mocking()
            return func(self, *args, **kwargs)

        def passthrough(self, *args, **kwargs):
            return func(self, *args, **kwargs)

        if _get_test_target():
            return passthrough
        else:
            return wrapped

def decorate(cls, func):
        return moto.mock_sqs(moto.mock_sns(responses.activate(func)))