# Copyright 2013 The Chromium Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. class RequestHeaders(object): '''A custom dictionary impementation for headers which ignores the case of requests, since different HTTP libraries seem to mangle them. ''' def __init__(self, dict_): if isinstance(dict_, RequestHeaders): self._dict = dict_ else: self._dict = dict((k.lower(), v) for k, v in dict_.iteritems()) def get(self, key, default=None): return self._dict.get(key.lower(), default) def __repr__(self): return repr(self._dict) def __str__(self): return repr(self._dict) class Request(object): '''Request data. ''' def __init__(self, path, host, headers): self.path = path.lstrip('/') self.host = host.rstrip('/') self.headers = RequestHeaders(headers) @staticmethod def ForTest(path, host=None, headers=None): return Request(path, host or 'http://developer.chrome.com', headers or {}) def __repr__(self): return 'Request(path=%s, host=%s, headers=%s)' % ( self.path, self.host, self.headers) def __str__(self): return repr(self) class _ContentBuilder(object): '''Builds the response content. ''' def __init__(self): self._buf = [] def Append(self, content): if isinstance(content, unicode): content = content.encode('utf-8', 'replace') self._buf.append(content) def ToString(self): self._Collapse() return self._buf[0] def __str__(self): return self.ToString() def __len__(self): return len(self.ToString()) def _Collapse(self): self._buf = [''.join(self._buf)] class Response(object): '''The response from Get(). ''' def __init__(self, content=None, headers=None, status=None): self.content = _ContentBuilder() if content is not None: self.content.Append(content) self.headers = {} if headers is not None: self.headers.update(headers) self.status = status @staticmethod def Ok(content, headers=None): '''Returns an OK (200) response. ''' return Response(content=content, headers=headers, status=200) @staticmethod def Redirect(url, permanent=False): '''Returns a redirect (301 or 302) response. ''' status = 301 if permanent else 302 return Response(headers={'Location': url}, status=status) @staticmethod def NotFound(content, headers=None): '''Returns a not found (404) response. ''' return Response(content=content, headers=headers, status=404) @staticmethod def NotModified(content, headers=None): return Response(content=content, headers=headers, status=304) @staticmethod def InternalError(content, headers=None): '''Returns an internal error (500) response. ''' return Response(content=content, headers=headers, status=500) def Append(self, content): '''Appends |content| to the response content. ''' self.content.append(content) def AddHeader(self, key, value): '''Adds a header to the response. ''' self.headers[key] = value def AddHeaders(self, headers): '''Adds several headers to the response. ''' self.headers.update(headers) def SetStatus(self, status): self.status = status def GetRedirect(self): if self.headers.get('Location') is None: return (None, None) return (self.headers.get('Location'), self.status == 301) def IsNotFound(self): return self.status == 404 def __eq__(self, other): return (isinstance(other, self.__class__) and str(other.content) == str(self.content) and other.headers == self.headers and other.status == self.status) def __ne__(self, other): return not (self == other) def __repr__(self): return 'Response(content=%s bytes, status=%s, headers=%s)' % ( len(self.content), self.status, self.headers) def __str__(self): return repr(self) class Servlet(object): def __init__(self, request): self._request = request def Get(self): '''Returns a Response. ''' raise NotImplemented()