Extensions Docs Server: Testing GithubFileSystem and cron jobs

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

 # Copyright (c) 2012 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.
 
 # This will attempt to import the actual App Engine modules, and if it fails,
 # they will be replaced with fake modules. This is useful during testing.
 try:
   import google.appengine.ext.blobstore as blobstore
   from google.appengine.ext.blobstore.blobstore import BlobReferenceProperty
   import google.appengine.ext.db as db
   import google.appengine.ext.webapp as webapp
   import google.appengine.api.files as files
   import google.appengine.api.memcache as memcache
   import google.appengine.api.urlfetch as urlfetch
 except ImportError:
   import re
+  from StringIO import StringIO
 
   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
@@ skipping 33 matching lines @@
       return _RPC()
 
     def make_fetch_call(self, rpc, url):
       rpc.result = self.fetch(url)
   urlfetch = FakeUrlFetch()
 
   class NotImplemented(object):
     def __getattr__(self, attr):
       raise NotImplementedError()
 
-  blobstore = NotImplemented()
-  files = NotImplemented()
+  _BLOBS = {}
+  class FakeBlobstore(object):
+    class BlobReader(object):
+      def __init__(self, blob_key):
+        self._data = _BLOBS[blob_key].getvalue()
+
+      def read(self):
+        return self._data
+
+  blobstore = FakeBlobstore()
+
+  class FakeFileInterface(object):
+    """This class allows a StringIO object to be used in a with block like a
+    file.
+    """
+    def __init__(self, io):
+      self._io = io
+
+    def __exit__(self, *args):
+      pass
+
+    def write(self, data):
+      self._io.write(data)
+
+    def __enter__(self, *args):
+      return self._io
+
+  class FakeFiles(object):
+    _next_blobstore_key = 0
+    class blobstore(object):
+      @staticmethod
+      def create():
+        FakeFiles._next_blobstore_key += 1
+        return FakeFiles._next_blobstore_key
+
+      @staticmethod
+      def get_blob_key(filename):
+        return filename
+
+    def open(self, filename, mode):
+      _BLOBS[filename] = StringIO()
+      return FakeFileInterface(_BLOBS[filename])
+
+    def GetBlobKeys(self):
+      return _BLOBS.keys()
+
+    def finalize(self, filename):
+      pass
+
+  files = FakeFiles()
 
   class InMemoryMemcache(object):
     """A fake memcache that does nothing.
     """
     class Client(object):
       def set_multi_async(self, mapping, namespace='', time=0):
         return
 
       def get_multi_async(self, keys, namespace='', time=0):
         return _RPC(result=dict((k, None) for k in keys))
 
     def set(self, key, value, namespace='', time=0):
       return
 
     def get(self, key, namespace='', time=0):
       return None
 
     def delete(self, key, namespace):
       return
+
   memcache = InMemoryMemcache()
 
   class webapp(object):
     class RequestHandler(object):
       """A fake webapp.RequestHandler class for Handler to extend.
       """
       def __init__(self, request, response):
         self.request = request
         self.response = response
 
       def redirect(self, path):
         self.request.path = path
 
   class _Db_Result(object):
+    def __init__(self, data):
+      self._data = data
+
+    class _Result(object):
+      def __init__(self, value):
+        self.value = value
+
     def get(self):
-      return []
+      return self._Result(self._data)
 
   class db(object):
+    _store = {}
     class StringProperty(object):
       pass
 
     class Model(object):
+      def __init__(self, key_='', value=''):
+        self._key = key_
+        self._value = value
+
       @staticmethod
-      def gql(*args):
-        return _Db_Result()
+      def gql(query, key):
+        return _Db_Result(db._store.get(key, None))
+
+      def put(self):
+        db._store[self._key] = self._value
 
   class BlobReferenceProperty(object):
     pass