Docserver overhaul: Gitiles away from me.

chrome/common/extensions/docs/server2/url_fetcher_fake.py

+# Copyright 2015 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.
+
+import logging
+import re
+import time
+
+from environment import IsAppEngine
+from future import Future
+from url_fetcher import UrlFetcher
+
+
+FAKE_URL_FETCHER_CONFIGURATION = None
+
+
+def ConfigureFakeUrlFetch(configuration):
+  """|configuration| is a dictionary mapping strings to fake urlfetch classes.
+  A fake urlfetch class just needs to have a fetch method. The keys of the
+  dictionary are treated as regex, and they are matched with the URL to
+  determine which fake urlfetch is used.
+  """
+  global FAKE_URL_FETCHER_CONFIGURATION
+  FAKE_URL_FETCHER_CONFIGURATION = dict(
+      (re.compile(k), v) for k, v in configuration.iteritems())
+
+
+def _GetConfiguration(key):
+  if not FAKE_URL_FETCHER_CONFIGURATION:
+    raise ValueError('No fake fetch paths have been configured. '
+                     'See ConfigureFakeUrlFetch in url_fetcher_fake.py.')
+  for k, v in FAKE_URL_FETCHER_CONFIGURATION.iteritems():
+    if k.match(key):
+      return v
+  raise ValueError('No configuration found for %s' % key)
+
+
+class UrlFetcherFake(UrlFetcher):
+  """A fake UrlFetcher implementation which may be configured with manual URL
+  overrides for testing. By default this 404s on everything.
+  """
+  class DownloadError(Exception):
+    pass
+
+  class _Response(object):
+    def __init__(self, content):
+      self.content = content
+      self.headers = {'Content-Type': 'none'}
+      self.status_code = 200
+
+  def FetchImpl(self, url, headers):
+    if IsAppEngine():
+      raise ValueError('Attempted to fetch URL from AppEngine: %s' % url)
+
+    url = url.split('?', 1)[0]
+    response = self._Response(_GetConfiguration(url).fetch(url))
+    if response.content is None:
+      response.status_code = 404
+    return response
+
+  def FetchAsyncImpl(self, url, headers):
+    return Future(callback=lambda: self.FetchImpl(url, headers))