Class HttpMockSequence

googleapiclient/http.py:1765–1826

Mock of httplib2.Http Mocks a sequence of calls to request returning different responses for each call. Create an instance initialized with the desired response headers and content and then use as if an httplib2.Http instance. http = HttpMockSequence([ ({'status': '401'},

class HttpMockSequence(object):
    """Mock of httplib2.Http

    Mocks a sequence of calls to request returning different responses for each
    call. Create an instance initialized with the desired response headers
    and content and then use as if an httplib2.Http instance.

      http = HttpMockSequence([
        ({'status': '401'}, ''),
        ({'status': '200'}, '{"access_token":"1/3w","expires_in":3600}'),
        ({'status': '200'}, 'echo_request_headers'),
        ])
      resp, content = http.request("http://examples.com")

    There are special values you can pass in for content to trigger
    behavours that are helpful in testing.

    'echo_request_headers' means return the request headers in the response body
    'echo_request_headers_as_json' means return the request headers in
       the response body
    'echo_request_body' means return the request body in the response body
    'echo_request_uri' means return the request uri in the response body
    """

    def __init__(self, iterable):
        """
        Args:
          iterable: iterable, a sequence of pairs of (headers, body)
        """
        self._iterable = iterable
        self.follow_redirects = True
        self.request_sequence = list()

    def request(
        self,
        uri,
        method="GET",
        body=None,
        headers=None,
        redirections=1,
        connection_type=None,
    ):
        # Remember the request so after the fact this mock can be examined
        self.request_sequence.append((uri, method, body, headers))
        resp, content = self._iterable.pop(0)
        if isinstance(content, str):
            content = content.encode("utf-8")

        if content == b"echo_request_headers":
            content = headers
        elif content == b"echo_request_headers_as_json":
            content = json.dumps(headers)
        elif content == b"echo_request_body":
            if hasattr(body, "read"):
                content = body.read()
            else:
                content = body
        elif content == b"echo_request_uri":