Imported Upstream version 1.11

5 files changed, 136 insertions, 0 deletions

diff --git a/test/appengine/__init__.py b/test/appengine/__init__.py
new file mode 100644
index 0000000..917544d
--- /dev/null
+++ b/test/appengine/__init__.py
@@ -0,0 +1,71 @@
+import os
+import sys
+import unittest
+from nose.plugins.skip import SkipTest
+
+
+def activate_sandbox():
+    """
+    Enables parts of the GAE sandbox that are relevant.
+
+    Inserts the stub module import hook which causes the usage of appengine-specific
+    httplib, httplib2, socket, etc.
+    """
+    from google.appengine.tools.devappserver2.python import sandbox
+
+    for name in list(sys.modules):
+        if name in sandbox.dist27.MODULE_OVERRIDES:
+            del sys.modules[name]
+    sys.meta_path.insert(0, sandbox.StubModuleImportHook())
+    sys.path_importer_cache = {}
+
+
+def deactivate_sandbox():
+    from google.appengine.tools.devappserver2.python import sandbox
+
+    sys.meta_path = [
+        x for x in sys.meta_path if not isinstance(x, sandbox.StubModuleImportHook)]
+    sys.path_importer_cache = {}
+
+    # Delete any instances of sandboxed modules.
+    for name in list(sys.modules):
+        if name in sandbox.dist27.MODULE_OVERRIDES:
+            del sys.modules[name]
+
+
+class AppEngineSandboxTest(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+
+        if sys.version_info[:2] != (2, 7):
+            raise SkipTest("App Engine only tests on py2.7")
+
+        if 'APPLICATION_ID' not in os.environ:
+            raise SkipTest("NoseGAE plugin not used.")
+
+        try:
+            activate_sandbox()
+        except ImportError:
+            raise SkipTest("App Engine SDK not available.")
+
+    @classmethod
+    def tearDownClass(self):
+        try:
+            deactivate_sandbox()
+        except ImportError:
+            pass
+
+
+class MockResponse(object):
+    def __init__(self, content, status_code, content_was_truncated, final_url, headers):
+        import httplib
+        from StringIO import StringIO
+
+        self.content = content
+        self.status_code = status_code
+        self.content_was_truncated = content_was_truncated
+        self.final_url = final_url
+        self.header_msg = httplib.HTTPMessage(StringIO(''.join(
+            ["%s: %s\n" % (k, v) for k, v in headers.iteritems()] + ["\n"])))
+        self.headers = self.header_msg.items()
diff --git a/test/appengine/app.yaml b/test/appengine/app.yaml
new file mode 100644
index 0000000..907c57f
--- /dev/null
+++ b/test/appengine/app.yaml
@@ -0,0 +1,11 @@
+# dummy app.yaml for nosegae
+
+api_version: 1
+runtime: python27
+threadsafe: true
+
+handlers:
+- url: /
+  static_files: README.md
+  upload: README.md
+  mime_type: text/plain
diff --git a/test/appengine/nose.cfg b/test/appengine/nose.cfg
new file mode 100644
index 0000000..8d8b3f1
--- /dev/null
+++ b/test/appengine/nose.cfg
@@ -0,0 +1,4 @@
+[nosetests]
+cover-min-percentage=0
+with-gae=1
+gae-application=test/appengine/app.yaml
diff --git a/test/appengine/requirements.txt b/test/appengine/requirements.txt
new file mode 100644
index 0000000..b6d79e0
--- /dev/null
+++ b/test/appengine/requirements.txt
@@ -0,0 +1 @@
+NoseGAE==0.5.7
diff --git a/test/appengine/test_urlfetch.py b/test/appengine/test_urlfetch.py
new file mode 100644
index 0000000..3f72023
--- /dev/null
+++ b/test/appengine/test_urlfetch.py
@@ -0,0 +1,49 @@
+from . import AppEngineSandboxTest, MockResponse
+
+from mock import patch
+from nose.plugins.skip import SkipTest
+from ..test_no_ssl import TestWithoutSSL
+
+
+class TestHTTP(AppEngineSandboxTest, TestWithoutSSL):
+    nosegae_urlfetch = True
+
+    def test_urlfetch_called_with_http(self):
+        """
+        Check that URLFetch is used to fetch non-https resources
+        """
+        resp = MockResponse(
+            'OK',
+            200,
+            False,
+            'http://www.google.com',
+            {'content-type': 'text/plain'})
+        with patch('google.appengine.api.urlfetch.fetch', return_value=resp) as fetchmock:
+            import urllib3
+            pool = urllib3.HTTPConnectionPool('www.google.com', '80')
+            r = pool.request('GET', '/')
+            self.assertEqual(r.status, 200, r.data)
+            self.assertEqual(fetchmock.call_count, 1)
+
+
+class TestHTTPS(AppEngineSandboxTest):
+    nosegae_urlfetch = True
+
+    def test_urlfetch_called_with_https(self):
+        """
+        Check that URLFetch is used when fetching https resources
+        """
+        raise SkipTest()  # Skipped for now because it fails.
+        resp = MockResponse(
+            'OK',
+            200,
+            False,
+            'https://www.google.com',
+            {'content-type': 'text/plain'})
+        with patch('google.appengine.api.urlfetch.fetch', return_value=resp) as fetchmock:
+            import urllib3
+            pool = urllib3.HTTPSConnectionPool('www.google.com', '443')
+            pool.ConnectionCls = urllib3.connection.UnverifiedHTTPSConnection
+            r = pool.request('GET', '/')
+            self.assertEqual(r.status, 200, r.data)
+            self.assertEqual(fetchmock.call_count, 1)