Upstream version 7.36.149.0

[platform/framework/web/crosswalk.git] / src / chrome / common / extensions / docs / server2 / fake_url_fetcher.py

diff --git a/src/chrome/common/extensions/docs/server2/fake_url_fetcher.py b/src/chrome/common/extensions/docs/server2/fake_url_fetcher.py
index d3a2fdc..dbfaa55 100644 (file)
--- a/src/chrome/common/extensions/docs/server2/fake_url_fetcher.py
+++ b/src/chrome/common/extensions/docs/server2/fake_url_fetcher.py
@@ -2,17 +2,17 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
-from functools import partial
 import os
+import posixpath
 
-from future import Gettable, Future
-from local_file_system import LocalFileSystem
+from future import Future
+from path_util import AssertIsDirectory, IsDirectory
 
 
 class _Response(object):
   def __init__(self, content=''):
     self.content = content
-    self.headers = { 'content-type': 'none' }
+    self.headers = {'Content-Type': 'none'}
     self.status_code = 200
 
 
@@ -25,8 +25,13 @@ class FakeUrlFetcher(object):
     self._async_resolve_count = 0
 
   def _ReadFile(self, filename):
-    with open(os.path.join(self._base_path, filename), 'r') as f:
-      return f.read()
+    # Fake DownloadError, the error that appengine usually raises.
+    class DownloadError(Exception): pass
+    try:
+      with open(os.path.join(self._base_path, filename), 'r') as f:
+        return f.read()
+    except IOError as e:
+      raise DownloadError(e)
 
   def _ListDir(self, directory):
     # In some tests, we need to test listing a directory from the HTML returned
@@ -51,7 +56,7 @@ class FakeUrlFetcher(object):
     def resolve():
       self._async_resolve_count += 1
       return self._DoFetch(url)
-    return Future(delegate=Gettable(resolve))
+    return Future(callback=resolve)
 
   def Fetch(self, url):
     self._sync_count += 1
@@ -60,7 +65,7 @@ class FakeUrlFetcher(object):
   def _DoFetch(self, url):
     url = url.rsplit('?', 1)[0]
     result = _Response()
-    if url.endswith('/'):
+    if IsDirectory(url):
       result.content = self._ListDir(url)
     else:
       result.content = self._ReadFile(url)
@@ -93,15 +98,9 @@ class FakeURLFSFetcher(object):
   '''Use a file_system to resolve fake fetches. Mimics the interface of Google
   Appengine's urlfetch.
   '''
-  @staticmethod
-  def Create(file_system):
-    return partial(FakeURLFSFetcher, file_system)
-
-  @staticmethod
-  def CreateLocal():
-    return partial(FakeURLFSFetcher, LocalFileSystem(''))
 
   def __init__(self, file_system, base_path):
+    AssertIsDirectory(base_path)
     self._base_path = base_path
     self._file_system = file_system
 
@@ -110,4 +109,50 @@ class FakeURLFSFetcher(object):
 
   def Fetch(self, url, **kwargs):
     return _Response(self._file_system.ReadSingle(
-        self._base_path + '/' + url, binary=True).Get())
+        posixpath.join(self._base_path, url)).Get())
+
+  def UpdateFS(self, file_system, base_path=None):
+    '''Replace the underlying FileSystem used to reslove URLs.
+    '''
+    self._file_system = file_system
+    self._base_path = base_path or self._base_path
+
+
+class MockURLFetcher(object):
+  def __init__(self, fetcher):
+    self._fetcher = fetcher
+    self.Reset()
+
+  def Fetch(self, url, **kwargs):
+    self._fetch_count += 1
+    return self._fetcher.Fetch(url, **kwargs)
+
+  def FetchAsync(self, url, **kwargs):
+    self._fetch_async_count += 1
+    future = self._fetcher.FetchAsync(url, **kwargs)
+    def resolve():
+      self._fetch_resolve_count += 1
+      return future.Get()
+    return Future(callback=resolve)
+
+  def CheckAndReset(self,
+                    fetch_count=0,
+                    fetch_async_count=0,
+                    fetch_resolve_count=0):
+    errors = []
+    for desc, expected, actual in (
+        ('fetch_count', fetch_count, self._fetch_count),
+        ('fetch_async_count', fetch_async_count, self._fetch_async_count),
+        ('fetch_resolve_count', fetch_resolve_count,
+                                self._fetch_resolve_count)):
+      if actual != expected:
+        errors.append('%s: expected %s got %s' % (desc, expected, actual))
+    try:
+      return (len(errors) == 0, ', '.join(errors))
+    finally:
+      self.Reset()
+
+  def Reset(self):
+    self._fetch_count = 0
+    self._fetch_async_count = 0
+    self._fetch_resolve_count = 0