Remove support for Google App Engine Standard
The 'URLFetch' service is only available on Python 2.7
and since 2.7 is no longer supported we can drop
urllib3's third-party module for App Engine as well
MODIFY
.travis.yml
.travis.yml
@@ -47,8 +47,6 @@ jobs:
env: NOX_SESSION=test-pypy
# Extras
- - python: 3.7
- env: NOX_SESSION=app_engine GAE_SDK_PATH=${HOME}/.cache/google_appengine
- python: 3.7
env: NOX_SESSION=google_brotli

DELETE
docs/reference/contrib/appengine.rst
-Google App Engine
-=================
-
-.. automodule:: urllib3.contrib.appengine
- :members:
- :undoc-members:
- :show-inheritance:

MODIFY
docs/reference/contrib/index.rst
docs/reference/contrib/index.rst
@@ -6,7 +6,6 @@ prime time or that require optional third-party dependencies.
.. toctree::
- appengine
ntlmpool
pyopenssl
securetransport

MODIFY
noxfile.py
noxfile.py
@@ -79,26 +79,6 @@ def google_brotli(session):
tests_impl(session, extras="socks,secure")
-@nox.session(python="3.7")
-def app_engine(session):
- session.install("-r", "dev-requirements.txt")
- session.install(".")
- session.run(
- "coverage",
- "run",
- "--parallel-mode",
- "-m",
- "pytest",
- "-r",
- "sx",
- "test/appengine",
- *session.posargs,
- )
- session.run("coverage", "combine")
- session.run("coverage", "report", "-m")
- session.run("coverage", "xml")
-
-
@nox.session()
def format(session):
"""Run code formatters."""

DELETE
src/urllib3/contrib/_appengine_environ.py
-"""
-This module provides means to detect the App Engine environment.
-"""
-
-import os
-
-
-def is_appengine():
- return is_local_appengine() or is_prod_appengine()
-
-
-def is_appengine_sandbox():
- """Reports if the app is running in the first generation sandbox.
-
- The second generation runtimes are technically still in a sandbox, but it
- is much less restrictive, so generally you shouldn't need to check for it.
- see https://cloud.google.com/appengine/docs/standard/runtimes
- """
- return is_appengine() and os.environ["APPENGINE_RUNTIME"] == "python27"
-
-
-def is_local_appengine():
- return "APPENGINE_RUNTIME" in os.environ and os.environ.get(
- "SERVER_SOFTWARE", ""
- ).startswith("Development/")
-
-
-def is_prod_appengine():
- return "APPENGINE_RUNTIME" in os.environ and os.environ.get(
- "SERVER_SOFTWARE", ""
- ).startswith("Google App Engine/")
-
-
-def is_prod_appengine_mvms():
- """Deprecated."""
- return False

DELETE
src/urllib3/contrib/appengine.py
-"""
-This module provides a pool manager that uses Google App Engine's
-`URLFetch Service <https://cloud.google.com/appengine/docs/python/urlfetch>`_.
-
-Example usage::
-
- from urllib3 import PoolManager
- from urllib3.contrib.appengine import AppEngineManager, is_appengine_sandbox
-
- if is_appengine_sandbox():
- # AppEngineManager uses AppEngine's URLFetch API behind the scenes
- http = AppEngineManager()
- else:
- # PoolManager uses a socket-level API behind the scenes
- http = PoolManager()
-
- r = http.request('GET', 'https://google.com/')
-
-There are `limitations <https://cloud.google.com/appengine/docs/python/\
-urlfetch/#Python_Quotas_and_limits>`_ to the URLFetch service and it may not be
-the best choice for your application. There are three options for using
-urllib3 on Google App Engine:
-
-1. You can use :class:`AppEngineManager` with URLFetch. URLFetch is
- cost-effective in many circumstances as long as your usage is within the
- limitations.
-2. You can use a normal :class:`~urllib3.PoolManager` by enabling sockets.
- Sockets also have `limitations and restrictions
- <https://cloud.google.com/appengine/docs/python/sockets/\
- #limitations-and-restrictions>`_ and have a lower free quota than URLFetch.
- To use sockets, be sure to specify the following in your ``app.yaml``::
-
- env_variables:
- GAE_USE_SOCKETS_HTTPLIB : 'true'
-
-3. If you are using `App Engine Flexible
-<https://cloud.google.com/appengine/docs/flexible/>`_, you can use the standard
-:class:`PoolManager` without any configuration or special environment variables.
-"""
-
-
-import io
-import logging
-import warnings
-from urllib.parse import urljoin
-
-from ..exceptions import (
- HTTPError,
- HTTPWarning,
- MaxRetryError,
- ProtocolError,
- SSLError,
- TimeoutError,
-)
-from ..request import RequestMethods
-from ..response import HTTPResponse
-from ..util.retry import Retry
-from ..util.timeout import Timeout
-from . import _appengine_environ
-
-try:
- from google.appengine.api import urlfetch
-except ImportError:
- urlfetch = None
-
-
-log = logging.getLogger(__name__)
-
-
-class AppEnginePlatformWarning(HTTPWarning):
- pass
-
-
-class AppEnginePlatformError(HTTPError):
- pass
-
-
-class AppEngineManager(RequestMethods):
- """
- Connection manager for Google App Engine sandbox applications.
-
- This manager uses the URLFetch service directly instead of using the
- emulated httplib, and is subject to URLFetch limitations as described in
- the App Engine documentation `here
- <https://cloud.google.com/appengine/docs/python/urlfetch>`_.
-
- Notably it will raise an :class:`AppEnginePlatformError` if:
- * URLFetch is not available.
- * If you attempt to use this on App Engine Flexible, as full socket
- support is available.
- * If a request size is more than 10 megabytes.
- * If a response size is more than 32 megabytes.
- * If you use an unsupported request method such as OPTIONS.
-
- Beyond those cases, it will raise normal urllib3 errors.
- """
-
- def __init__(
- self,
- headers=None,
- retries=None,
- validate_certificate=True,
- urlfetch_retries=True,
- ):
- if not urlfetch:
- raise AppEnginePlatformError(
- "URLFetch is not available in this environment."
- )
-
- warnings.warn(
- "urllib3 is using URLFetch on Google App Engine sandbox instead "
- "of sockets. To use sockets directly instead of URLFetch see "
- "https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
- AppEnginePlatformWarning,
- )
-
- RequestMethods.__init__(self, headers)
- self.validate_certificate = validate_certificate
- self.urlfetch_retries = urlfetch_retries
-
- self.retries = retries or Retry.DEFAULT
-
- def __enter__(self):
- return self
-
- def __exit__(self, exc_type, exc_val, exc_tb):
- # Return False to re-raise any potential exceptions
- return False
-
- def urlopen(
- self,
- method,
- url,
- body=None,
- headers=None,
- retries=None,
- redirect=True,
- timeout=Timeout.DEFAULT_TIMEOUT,
- **response_kw,
- ):
-
- retries = self._get_retries(retries, redirect)
-
- try:
- follow_redirects = redirect and retries.redirect != 0 and retries.total
- response = urlfetch.fetch(
- url,
- payload=body,
- method=method,
- headers=headers or {},
- allow_truncated=False,
- follow_redirects=self.urlfetch_retries and follow_redirects,
- deadline=self._get_absolute_timeout(timeout),
- validate_certificate=self.validate_certificate,
- )
- except urlfetch.DeadlineExceededError as e:
- raise TimeoutError(self, e)
-
- except urlfetch.InvalidURLError as e:
- if "too large" in str(e):
- raise AppEnginePlatformError(
- "URLFetch request too large, URLFetch only "
- "supports requests up to 10mb in size.",
- e,
- )
- raise ProtocolError(e)
-
- except urlfetch.DownloadError as e:
- if "Too many redirects" in str(e):
- raise MaxRetryError(self, url, reason=e)
- raise ProtocolError(e)
-
- except urlfetch.ResponseTooLargeError as e:
- raise AppEnginePlatformError(
- "URLFetch response too large, URLFetch only supports"
- "responses up to 32mb in size.",
- e,
- )
-
- except urlfetch.SSLCertificateError as e:
- raise SSLError(e)
-
- except urlfetch.InvalidMethodError as e:
- raise AppEnginePlatformError(
- f"URLFetch does not support method: {method}", e
- )
-
- http_response = self._urlfetch_response_to_http_response(
- response, retries=retries, **response_kw
- )
-
- # Handle redirect?
- redirect_location = redirect and http_response.get_redirect_location()
- if redirect_location:
- # Check for redirect response
- if self.urlfetch_retries and retries.raise_on_redirect:
- raise MaxRetryError(self, url, "too many redirects")
- else:
- if http_response.status == 303:
- method = "GET"
-
- try:
- retries = retries.increment(
- method, url, response=http_response, _pool=self
- )
- except MaxRetryError:
- if retries.raise_on_redirect:
- raise MaxRetryError(self, url, "too many redirects")
- return http_response
-
- retries.sleep_for_retry(http_response)
- log.debug("Redirecting %s -> %s", url, redirect_location)
- redirect_url = urljoin(url, redirect_location)
- return self.urlopen(
- method,
- redirect_url,
- body,
- headers,
- retries=retries,
- redirect=redirect,
- timeout=timeout,
- **response_kw,
- )
-
- # Check if we should retry the HTTP response.
- has_retry_after = bool(http_response.getheader("Retry-After"))
- if retries.is_retry(method, http_response.status, has_retry_after):
- retries = retries.increment(method, url, response=http_response, _pool=self)
- log.debug("Retry: %s", url)
- retries.sleep(http_response)
- return self.urlopen(
- method,
- url,
- body=body,
- headers=headers,
- retries=retries,
- redirect=redirect,
- timeout=timeout,
- **response_kw,
- )
-
- return http_response
-
- def _urlfetch_response_to_http_response(self, urlfetch_resp, **response_kw):
-
- if is_prod_appengine():
- # Production GAE handles deflate encoding automatically, but does
- # not remove the encoding header.
- content_encoding = urlfetch_resp.headers.get("content-encoding")
-
- if content_encoding == "deflate":
- del urlfetch_resp.headers["content-encoding"]
-
- transfer_encoding = urlfetch_resp.headers.get("transfer-encoding")
- # We have a full response's content,
- # so let's make sure we don't report ourselves as chunked data.
- if transfer_encoding == "chunked":
- encodings = transfer_encoding.split(",")
- encodings.remove("chunked")
- urlfetch_resp.headers["transfer-encoding"] = ",".join(encodings)
-
- original_response = HTTPResponse(
- # In order for decoding to work, we must present the content as
- # a file-like object.
- body=io.BytesIO(urlfetch_resp.content),
- msg=urlfetch_resp.header_msg,
- headers=urlfetch_resp.headers,
- status=urlfetch_resp.status_code,
- **response_kw,
- )
-
- return HTTPResponse(
- body=io.BytesIO(urlfetch_resp.content),
- headers=urlfetch_resp.headers,
- status=urlfetch_resp.status_code,
- original_response=original_response,
- **response_kw,
- )
-
- def _get_absolute_timeout(self, timeout):
- if timeout is Timeout.DEFAULT_TIMEOUT:
- return None # Defer to URLFetch's default.
- if isinstance(timeout, Timeout):
- if timeout._read is not None or timeout._connect is not None:
- warnings.warn(
- "URLFetch does not support granular timeout settings, "
- "reverting to total or default URLFetch timeout.",
- AppEnginePlatformWarning,
- )
- return timeout.total
- return timeout
-
- def _get_retries(self, retries, redirect):
- if not isinstance(retries, Retry):
- retries = Retry.from_int(retries, redirect=redirect, default=self.retries)
-
- if retries.connect or retries.read or retries.redirect:
- warnings.warn(
- "URLFetch only supports total retries and does not "
- "recognize connect, read, or redirect retry parameters.",
- AppEnginePlatformWarning,
- )
-
- return retries
-
-
-# Alias methods from _appengine_environ to maintain public API interface.
-
-is_appengine = _appengine_environ.is_appengine
-is_appengine_sandbox = _appengine_environ.is_appengine_sandbox
-is_local_appengine = _appengine_environ.is_local_appengine
-is_prod_appengine = _appengine_environ.is_prod_appengine
-is_prod_appengine_mvms = _appengine_environ.is_prod_appengine_mvms

MODIFY
src/urllib3/util/connection.py
src/urllib3/util/connection.py
@@ -2,8 +2,7 @@ import socket
from urllib3.exceptions import LocationParseError
-from ..contrib import _appengine_environ
-from .wait import NoWayToWaitForSocketError, wait_for_read
+from .wait import wait_for_read
def is_connection_dropped(conn): # Platform-specific
@@ -17,15 +16,10 @@ def is_connection_dropped(conn): # Platform-specific
let the platform handle connection recycling transparently for us.
"""
sock = getattr(conn, "sock", False)
- if sock is False: # Platform-specific: AppEngine
- return False
if sock is None: # Connection already closed (such as by httplib).
return True
- try:
# Returns True if readable, which here means it's been dropped
return wait_for_read(sock, timeout=0.0)
- except NoWayToWaitForSocketError: # Platform-specific: AppEngine
- return False
# This function is copied from socket.py in the Python 2.7 standard
@@ -117,13 +111,6 @@ def _has_ipv6(host):
sock = None
has_ipv6 = False
- # App Engine doesn't support IPV6 sockets and actually has a quota on the
- # number of sockets that can be used, so just early out here instead of
- # creating a socket needlessly.
- # See https://github.com/urllib3/urllib3/issues/1446
- if _appengine_environ.is_appengine_sandbox():
- return False
-
if socket.has_ipv6:
# has_ipv6 returns true if cPython was compiled with IPv6 support.
# It does not tell us if the system has IPv6 support enabled. To

MODIFY
src/urllib3/util/response.py
src/urllib3/util/response.py
@@ -99,7 +99,4 @@ def is_response_to_head(response):
used 'HEAD' as a method.
"""
# FIXME: Can we do this somehow without accessing private httplib _method?
- method = response._method
- if isinstance(method, int): # Platform-specific: Appengine
- return method == 3
- return method.upper() == "HEAD"
+ return response._method.upper() == "HEAD"

MODIFY
src/urllib3/util/wait.py
src/urllib3/util/wait.py
import select
from functools import partial
-__all__ = ["NoWayToWaitForSocketError", "wait_for_read", "wait_for_write"]
-
-
-class NoWayToWaitForSocketError(Exception):
- pass
+__all__ = ["wait_for_read", "wait_for_write"]
# How should we wait on sockets?
@@ -70,10 +66,6 @@ def poll_wait_for_socket(sock, read=False, write=False, timeout=None):
return bool(do_poll(timeout))
-def null_wait_for_socket(*args, **kwargs):
- raise NoWayToWaitForSocketError("no select-equivalent available")
-
-
def _have_working_poll():
# Apparently some systems have a select.poll that fails as soon as you try
# to use it, either due to strange configuration or broken monkeypatching
@@ -97,8 +89,6 @@ def wait_for_socket(*args, **kwargs):
wait_for_socket = poll_wait_for_socket
elif hasattr(select, "select"):
wait_for_socket = select_wait_for_socket
- else: # Platform-specific: Appengine.
- wait_for_socket = null_wait_for_socket
return wait_for_socket(*args, **kwargs)

DELETE
test/appengine/__init__.py
test/appengine/__init__.py
DELETE
test/appengine/conftest.py
-# Copyright 2015 Google Inc. All rights reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import os
-import sys
-
-# Import py.test hooks and fixtures for App Engine
-try:
- from gcp_devrel.testing.appengine import (
- pytest_configure,
- pytest_runtest_call,
- testbed,
- )
-except ImportError:
- pass
-
-import pytest
-
-__all__ = [
- "pytest_configure",
- "pytest_runtest_call",
- "pytest_ignore_collect",
- "testbed",
- "sandbox",
-]
-
-
-@pytest.fixture
-def sandbox(testbed):
- """
- 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.
- """
- try:
- from google.appengine.tools.devappserver2.python import sandbox
- except ImportError:
- from google.appengine.tools.devappserver2.python.runtime 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 = {}
-
- yield testbed
-
- 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]
-
-
-def pytest_ignore_collect(path, config):
- """Skip App Engine tests in python 3 or if no SDK is available."""
- if "appengine" in str(path):
- return True
- if not os.environ.get("GAE_SDK_PATH"):
- return True
- return False

DELETE
test/appengine/test_gae_manager.py
-from test import SHORT_TIMEOUT
-from test.with_dummyserver import test_connectionpool
-
-import pytest
-
-import dummyserver.testcase
-import urllib3.exceptions
-import urllib3.util.retry
-import urllib3.util.url
-from urllib3.contrib import appengine
-
-
-# This class is used so we can re-use the tests from the connection pool.
-# It proxies all requests to the manager.
-class MockPool:
- def __init__(self, host, port, manager, scheme="http"):
- self.host = host
- self.port = port
- self.manager = manager
- self.scheme = scheme
-
- def request(self, method, url, *args, **kwargs):
- url = self._absolute_url(url)
- return self.manager.request(method, url, *args, **kwargs)
-
- def urlopen(self, method, url, *args, **kwargs):
- url = self._absolute_url(url)
- return self.manager.urlopen(method, url, *args, **kwargs)
-
- def _absolute_url(self, path):
- return urllib3.util.url.Url(
- scheme=self.scheme, host=self.host, port=self.port, path=path
- ).url
-
-
-# Note that this doesn't run in the sandbox, it only runs with the URLFetch
-# API stub enabled. There's no need to enable the sandbox as we know for a fact
-# that URLFetch is used by the connection manager.
-@pytest.mark.usefixtures("testbed")
-class TestGAEConnectionManager(test_connectionpool.TestConnectionPool):
- def setup_method(self, method):
- self.manager = appengine.AppEngineManager()
- self.pool = MockPool(self.host, self.port, self.manager)
-
- # Tests specific to AppEngineManager
-
- def test_exceptions(self):
- # DeadlineExceededError -> TimeoutError
- with pytest.raises(urllib3.exceptions.TimeoutError):
- self.pool.request(
- "GET",
- f"/sleep?seconds={5 * SHORT_TIMEOUT}",
- timeout=SHORT_TIMEOUT,
- )
-
- # InvalidURLError -> ProtocolError
- with pytest.raises(urllib3.exceptions.ProtocolError):
- self.manager.request("GET", "ftp://invalid/url")
-
- # DownloadError -> ProtocolError
- with pytest.raises(urllib3.exceptions.ProtocolError):
- self.manager.request("GET", "http://0.0.0.0")
-
- # ResponseTooLargeError -> AppEnginePlatformError
- with pytest.raises(appengine.AppEnginePlatformError):
- self.pool.request(
- "GET", "/nbytes?length=33554433"
- ) # One byte over 32 megabytes.
-
- # URLFetch reports the request too large error as a InvalidURLError,
- # which maps to a AppEnginePlatformError.
- body = b"1" * 10485761 # One byte over 10 megabytes.
- with pytest.raises(appengine.AppEnginePlatformError):
- self.manager.request("POST", "/", body=body)
-
- # Re-used tests below this line.
- # Subsumed tests
- test_timeout_float = None # Covered by test_exceptions.
-
- # Non-applicable tests
- test_conn_closed = None
- test_nagle = None
- test_socket_options = None
- test_disable_default_socket_options = None
- test_defaults_are_applied = None
- test_tunnel = None
- test_keepalive = None
- test_keepalive_close = None
- test_connection_count = None
- test_connection_count_bigpool = None
- test_for_double_release = None
- test_release_conn_parameter = None
- test_stream_keepalive = None
- test_cleanup_on_connection_error = None
- test_read_chunked_short_circuit = None
- test_read_chunked_on_closed_response = None
-
- # Tests that should likely be modified for appengine specific stuff
- test_timeout = None
- test_connect_timeout = None
- test_connection_error_retries = None
- test_total_timeout = None
- test_none_total_applies_connect = None
- test_timeout_success = None
- test_source_address_error = None
- test_bad_connect = None
- test_partial_response = None
- test_dns_error = None
-
-
-@pytest.mark.usefixtures("testbed")
-class TestGAEConnectionManagerWithSSL(dummyserver.testcase.HTTPSDummyServerTestCase):
- def setup_method(self, method):
- self.manager = appengine.AppEngineManager()
- self.pool = MockPool(self.host, self.port, self.manager, "https")
-
- def test_exceptions(self):
- # SSLCertificateError -> SSLError
- # SSLError is raised with dummyserver because URLFetch doesn't allow
- # self-signed certs.
- with pytest.raises(urllib3.exceptions.SSLError):
- self.pool.request("GET", "/")
-
-
-@pytest.mark.usefixtures("testbed")
-class TestGAERetry(test_connectionpool.TestRetry):
- def setup_method(self, method):
- self.manager = appengine.AppEngineManager()
- self.pool = MockPool(self.host, self.port, self.manager)
-
- def test_default_method_whitelist_retried(self):
- """ urllib3 should retry methods in the default method whitelist """
- retry = urllib3.util.retry.Retry(total=1, status_forcelist=[418])
- # Use HEAD instead of OPTIONS, as URLFetch doesn't support OPTIONS
- resp = self.pool.request(
- "HEAD",
- "/successful_retry",
- headers={"test-name": "test_default_whitelist"},
- retries=retry,
- )
- assert resp.status == 200
-
- def test_retry_return_in_response(self):
- headers = {"test-name": "test_retry_return_in_response"}
- retry = urllib3.util.retry.Retry(total=2, status_forcelist=[418])
- resp = self.pool.request(
- "GET", "/successful_retry", headers=headers, retries=retry
- )
- assert resp.status == 200
- assert resp.retries.total == 1
- # URLFetch use absolute urls.
- assert resp.retries.history == (
- urllib3.util.retry.RequestHistory(
- "GET", self.pool._absolute_url("/successful_retry"), None, 418, None
- ),
- )
-
- # test_max_retry = None
- # test_disabled_retry = None
- # We don't need these tests because URLFetch resolves its own redirects.
- test_retry_redirect_history = None
- test_multi_redirect_history = None
-
-
-@pytest.mark.usefixtures("testbed")
-class TestGAERetryAfter(test_connectionpool.TestRetryAfter):
- def setup_method(self, method):
- # Disable urlfetch which doesn't respect Retry-After header.
- self.manager = appengine.AppEngineManager(urlfetch_retries=False)
- self.pool = MockPool(self.host, self.port, self.manager)
-
-
-def test_gae_environ():
- assert not appengine.is_appengine()
- assert not appengine.is_appengine_sandbox()
- assert not appengine.is_local_appengine()
- assert not appengine.is_prod_appengine()
- assert not appengine.is_prod_appengine_mvms()

DELETE
test/appengine/test_urlfetch.py
-"""These tests ensure that when running in App Engine standard with the
-App Engine sandbox enabled that urllib3 appropriately uses the App
-Engine-patched version of httplib to make requests."""
-
-from unittest.mock import patch
-
-import httplib
-import pytest
-import StringIO
-
-from ..test_no_ssl import TestWithoutSSL
-
-
-class MockResponse:
- def __init__(self, content, status_code, content_was_truncated, final_url, headers):
-
- 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.StringIO(
- "".join([f"{k}: {v}\n" for k, v in headers.iteritems()] + ["\n"])
- )
- )
- self.headers = headers
-
-
-@pytest.mark.usefixtures("sandbox")
-class TestHTTP(TestWithoutSSL):
- 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"}
- )
- fetch_patch = patch("google.appengine.api.urlfetch.fetch", return_value=resp)
- with fetch_patch as fetch_mock:
- import urllib3
-
- pool = urllib3.HTTPConnectionPool("www.google.com", "80")
- r = pool.request("GET", "/")
- assert r.status == 200, r.data
- assert fetch_mock.call_count == 1
-
-
-@pytest.mark.usefixtures("sandbox")
-class TestHTTPS:
- @pytest.mark.xfail(
- reason="This is not yet supported by urlfetch, presence of the ssl "
- "module will bypass urlfetch."
- )
- def test_urlfetch_called_with_https(self):
- """
- Check that URLFetch is used when fetching https resources
- """
- resp = MockResponse(
- "OK", 200, False, "https://www.google.com", {"content-type": "text/plain"}
- )
- fetch_patch = patch("google.appengine.api.urlfetch.fetch", return_value=resp)
- with fetch_patch as fetch_mock:
- import urllib3
-
- pool = urllib3.HTTPSConnectionPool("www.google.com", "443")
- pool.ConnectionCls = urllib3.connection.UnverifiedHTTPSConnection
- r = pool.request("GET", "/")
- assert r.status == 200, r.data
- assert fetch_mock.call_count == 1

MODIFY
test/test_util.py
test/test_util.py
@@ -708,13 +708,6 @@ class TestUtil:
instance.bind.return_value = True
assert _has_ipv6("::1")
- def test_has_ipv6_disabled_on_appengine(self):
- gae_patch = patch(
- "urllib3.contrib._appengine_environ.is_appengine_sandbox", return_value=True
- )
- with gae_patch:
- assert not _has_ipv6("::1")
-
def test_ip_family_ipv6_enabled(self):
with patch("urllib3.util.connection.HAS_IPV6", True):
assert allowed_gai_family() == socket.AF_UNSPEC

Remove support for Google App Engine Standard

The 'URLFetch' service is only available on Python 2.7
and since 2.7 is no longer supported we can drop
urllib3's third-party module for App Engine as well