Gecko [ www.smanegeri1koba.sch.id ]

Name	Size	Permission
testing	[ DIR ]	drwxr-xr-x
__init__.py	1.22 KB	-rw-r--r--
base_api.py	29.29 KB	-rw-r--r--
base_api_test.py	14.12 KB	-rw-r--r--
batch.py	17.93 KB	-rw-r--r--
batch_test.py	24.73 KB	-rw-r--r--
buffered_stream.py	2.52 KB	-rw-r--r--
buffered_stream_test.py	2.4 KB	-rw-r--r--
compression.py	5.02 KB	-rw-r--r--
compression_test.py	5.19 KB	-rw-r--r--
credentials_lib.py	29.42 KB	-rw-r--r--
credentials_lib_test.py	7.58 KB	-rw-r--r--
encoding.py	1.16 KB	-rw-r--r--
encoding_helper.py	30 KB	-rw-r--r--
encoding_test.py	31.97 KB	-rw-r--r--
exceptions.py	4.98 KB	-rw-r--r--
exceptions_test.py	2.88 KB	-rw-r--r--
extra_types.py	9.74 KB	-rw-r--r--
extra_types_test.py	7.81 KB	-rw-r--r--
gzip.py	21.39 KB	-rw-r--r--
gzip_test.py	17.17 KB	-rw-r--r--
http_wrapper.py	15.09 KB	-rw-r--r--
http_wrapper_test.py	5.09 KB	-rw-r--r--
list_pager.py	5.49 KB	-rw-r--r--
list_pager_test.py	13.08 KB	-rw-r--r--
stream_slice.py	2.67 KB	-rw-r--r--
stream_slice_test.py	2.2 KB	-rw-r--r--
transfer.py	45.42 KB	-rw-r--r--
transfer_test.py	26.06 KB	-rw-r--r--
util.py	8.28 KB	-rw-r--r--
util_test.py	7.01 KB	-rw-r--r--

Code Editor : http_wrapper_test.py

#
# Copyright 2015 Google Inc.
#
# 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.

"""Tests for http_wrapper."""
import socket
import unittest

import httplib2
from six.moves import http_client

from mock import patch

from apitools.base.py import exceptions
from apitools.base.py import http_wrapper

# pylint: disable=ungrouped-imports
try:
    from oauth2client.client import HttpAccessTokenRefreshError
    from oauth2client.client import AccessTokenRefreshError
    _TOKEN_REFRESH_STATUS_AVAILABLE = True
except ImportError:
    from oauth2client.client import AccessTokenRefreshError
    _TOKEN_REFRESH_STATUS_AVAILABLE = False

class _MockHttpRequest(object):

url = None

class _MockHttpResponse(object):

def __init__(self, status_code):
        self.response = {'status': status_code}

class RaisesExceptionOnLen(object):

"""Supports length property but raises if __len__ is used."""

def __len__(self):
        raise Exception('len() called unnecessarily')

def length(self):
        return 1

class HttpWrapperTest(unittest.TestCase):

def testRequestBodyUsesLengthProperty(self):
        http_wrapper.Request(body=RaisesExceptionOnLen())

def testRequestBodyWithLen(self):
        http_wrapper.Request(body='burrito')

@unittest.skipIf(not _TOKEN_REFRESH_STATUS_AVAILABLE,
                     'oauth2client<1.5 lacks HttpAccessTokenRefreshError.')
    def testExceptionHandlerHttpAccessTokenError(self):
        exception_arg = HttpAccessTokenRefreshError(status=503)
        retry_args = http_wrapper.ExceptionRetryArgs(
            http={'connections': {}}, http_request=_MockHttpRequest(),
            exc=exception_arg, num_retries=0, max_retry_wait=0,
            total_wait_sec=0)

# Disable time.sleep for this handler as it is called with
        # a minimum value of 1 second.
        with patch('time.sleep', return_value=None):
            http_wrapper.HandleExceptionsAndRebuildHttpConnections(
                retry_args)

@unittest.skipIf(not _TOKEN_REFRESH_STATUS_AVAILABLE,
                     'oauth2client<1.5 lacks HttpAccessTokenRefreshError.')
    def testExceptionHandlerHttpAccessTokenErrorRaises(self):
        exception_arg = HttpAccessTokenRefreshError(status=200)
        retry_args = http_wrapper.ExceptionRetryArgs(
            http={'connections': {}}, http_request=_MockHttpRequest(),
            exc=exception_arg, num_retries=0, max_retry_wait=0,
            total_wait_sec=0)

# Disable time.sleep for this handler as it is called with
        # a minimum value of 1 second.
        with self.assertRaises(HttpAccessTokenRefreshError):
            with patch('time.sleep', return_value=None):
                http_wrapper.HandleExceptionsAndRebuildHttpConnections(
                    retry_args)

def testExceptionHandlerAccessTokenErrorRaises(self):
        exception_arg = AccessTokenRefreshError()
        retry_args = http_wrapper.ExceptionRetryArgs(
            http={'connections': {}}, http_request=_MockHttpRequest(),
            exc=exception_arg, num_retries=0, max_retry_wait=0,
            total_wait_sec=0)

# Disable time.sleep for this handler as it is called with
        # a minimum value of 1 second.
        with self.assertRaises(AccessTokenRefreshError):
            with patch('time.sleep', return_value=None):
                http_wrapper.HandleExceptionsAndRebuildHttpConnections(
                    retry_args)

def testDefaultExceptionHandler(self):
        """Ensures exception handles swallows (retries)"""
        mock_http_content = 'content'.encode('utf8')
        for exception_arg in (
                http_client.BadStatusLine('line'),
                http_client.IncompleteRead('partial'),
                http_client.ResponseNotReady(),
                socket.error(),
                socket.gaierror(),
                httplib2.ServerNotFoundError(),
                ValueError(),
                exceptions.RequestError(),
                exceptions.BadStatusCodeError(
                    {'status': 503}, mock_http_content, 'url'),
                exceptions.RetryAfterError(
                    {'status': 429}, mock_http_content, 'url', 0)):

retry_args = http_wrapper.ExceptionRetryArgs(
                http={'connections': {}}, http_request=_MockHttpRequest(),
                exc=exception_arg, num_retries=0, max_retry_wait=0,
                total_wait_sec=0)

# Disable time.sleep for this handler as it is called with
            # a minimum value of 1 second.
            with patch('time.sleep', return_value=None):
                http_wrapper.HandleExceptionsAndRebuildHttpConnections(
                    retry_args)

${this.title}

Code Editor : http_wrapper_test.py