-
Notifications
You must be signed in to change notification settings - Fork 309
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add urllib3 AuthorizedHttp #19
Conversation
965389b
to
624fa02
Compare
from six.moves import http_client | ||
|
||
DEFAULT_REFRESH_STATUS_CODES = (http_client.UNAUTHORIZED,) | ||
DEFAULT_MAX_REFRESH_ATTEMPTS = 2 |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
try: | ||
import certifi | ||
except ImportError: # pragma: NO COVER | ||
certifi = None |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
http = urllib3.PoolManager() | ||
request = google.auth.transport.urllib3.Request(http) | ||
|
||
credentials.refresh(request) |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
|
||
|
||
def _make_default_http(): | ||
if certifi: |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
if certifi: | ||
return urllib3.PoolManager( | ||
cert_reqs='CERT_REQUIRED', | ||
ca_certs=certifi.where()) |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
self.headers = headers or {} | ||
|
||
def urlopen(self, method, url, body=None, headers=None, **kwargs): | ||
self.requests.append((method, url, body, headers)) |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
return self.responses.pop(0) | ||
|
||
|
||
class MockResponse(object): |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
mock.sentinel.credentials) | ||
|
||
assert authed_http.credentials == mock.sentinel.credentials | ||
assert authed_http.http is not None |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
assert authed_http.http is not None | ||
|
||
def test_urlopen_no_refresh(self): | ||
mock_credentials = mock.Mock(wraps=MockCredentials()) |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
def test_urlopen_refresh(self): | ||
mock_credentials = mock.Mock(wraps=MockCredentials()) | ||
mock_final_response = MockResponse(status=http_client.OK) | ||
# First request will 403, second request will succeed. |
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
This comment was marked as spam.
Sorry, something went wrong.
@jonparrott We have anything left? |
Don't think so. On Wed, Oct 12, 2016, 10:46 PM Danny Hermes [email protected]
|
👍 LGTM then |
Implementation of the plan detailed over at #15.