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

   1 # Copyright 2015 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 import logging
   6 import re
   7 import time
   8
   9 from environment import IsAppEngine
  10 from future import Future
  11 from url_fetcher import UrlFetcher
  12
  13
  14 FAKE_URL_FETCHER_CONFIGURATION = None
  15
  16
  17 def ConfigureFakeUrlFetch(configuration):
  18   """|configuration| is a dictionary mapping strings to fake urlfetch classes.
  19   A fake urlfetch class just needs to have a fetch method. The keys of the
  20   dictionary are treated as regex, and they are matched with the URL to
  21   determine which fake urlfetch is used.
  22   """
  23   global FAKE_URL_FETCHER_CONFIGURATION
  24   FAKE_URL_FETCHER_CONFIGURATION = dict(
  25       (re.compile(k), v) for k, v in configuration.iteritems())
  26
  27
  28 def _GetConfiguration(key):
  29   if not FAKE_URL_FETCHER_CONFIGURATION:
  30     raise ValueError('No fake fetch paths have been configured. '
  31                      'See ConfigureFakeUrlFetch in url_fetcher_fake.py.')
  32   for k, v in FAKE_URL_FETCHER_CONFIGURATION.iteritems():
  33     if k.match(key):
  34       return v
  35   raise ValueError('No configuration found for %s' % key)
  36
  37
  38 class UrlFetcherFake(UrlFetcher):
  39   """A fake UrlFetcher implementation which may be configured with manual URL
  40   overrides for testing. By default this 404s on everything.
  41   """
  42   class DownloadError(Exception):
  43     pass
  44
  45   class _Response(object):
  46     def __init__(self, content):
  47       self.content = content
  48       self.headers = {'Content-Type': 'none'}
  49       self.status_code = 200
  50
  51   def FetchImpl(self, url, headers):
  52     if IsAppEngine():
  53       raise ValueError('Attempted to fetch URL from AppEngine: %s' % url)
  54
  55     url = url.split('?', 1)[0]
  56     response = self._Response(_GetConfiguration(url).fetch(url))
  57     if response.content is None:
  58       response.status_code = 404
  59     return response
  60
  61   def FetchAsyncImpl(self, url, headers):
  62     return Future(callback=lambda: self.FetchImpl(url, headers))