# Copyright 2014 Google Inc. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """HTTP helpers mock functionality.""" from six.moves import http_client class ResponseMock(dict): """Mock HTTP response""" def __init__(self, vals=None): if vals is None: vals = {} self.update(vals) self.status = int(self.get('status', http_client.OK)) class HttpMock(object): """Mock of HTTP object.""" def __init__(self, headers=None, data=None): """HttpMock constructor. Args: headers: dict, header to return with response """ if headers is None: headers = {'status': http_client.OK} self.data = data self.response_headers = headers self.headers = None self.uri = None self.method = None self.body = None self.headers = None self.requests = 0 def request(self, uri, method='GET', body=None, headers=None, redirections=1, connection_type=None): self.uri = uri self.method = method self.body = body self.headers = headers self.redirections = redirections self.requests += 1 return ResponseMock(self.response_headers), self.data class HttpMockSequence(object): """Mock of HTTP object with multiple return values. 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 HttpMock instance:: http = HttpMockSequence([ ({'status': '401'}, b''), ({'status': '200'}, b'{"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_body' means return the request body in the response body """ def __init__(self, iterable): """HttpMockSequence constructor. Args: iterable: iterable, a sequence of pairs of (headers, body) """ self._iterable = iterable self.requests = [] def request(self, uri, method='GET', body=None, headers=None, redirections=1, connection_type=None): resp, content = self._iterable.pop(0) self.requests.append({ 'method': method, 'uri': uri, 'body': body, 'headers': headers, }) # Read any underlying stream before sending the request. body_stream_content = (body.read() if getattr(body, 'read', None) else None) if content == 'echo_request_headers': content = headers elif content == 'echo_request_body': content = (body if body_stream_content is None else body_stream_content) return ResponseMock(resp), content class CacheMock(object): def __init__(self): self.cache = {} def get(self, key, namespace=''): # ignoring namespace for easier testing return self.cache.get(key, None)