kennethreitz/requests3
1
0
Fork 0
mirror of https://github.com/kennethreitz/requests3.git synced 2026-06-05 23:10:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

that approach works

Browse Source
This commit is contained in:
Kenneth Reitz
2019-04-21 13:04:46 -04:00
parent b7427531ed
commit 5a1da4e326
7 changed files with 10 additions and 3421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
requests3/http_utils.py
+4 -4
View File
@@ -657,12 +657,12 @@ def should_bypass_proxies(url: str, no_proxy: typing.Optional[str]) -> bool:
no_proxy_arg = no_proxy
if no_proxy is None:
no_proxy = get_proxy("no_proxy")
netloc = urlparse(url).netloc
parsed = urlparse(url)
if no_proxy:
# We need to check whether we match here. We need to see if we match
# the end of the hostname, both with and without the port.
no_proxy = (host for host in no_proxy.replace(" ", "").split(",") if host)
ip = netloc.split(":")[0]
ip = parsed.netloc.split(":")[0]
if is_ipv4_address(ip):
for proxy_ip in no_proxy:
if is_valid_cidr(proxy_ip):
@@ -680,13 +680,13 @@ def should_bypass_proxies(url: str, no_proxy: typing.Optional[str]) -> bool:
host_with_port += ":{}".format(parsed.port)
for host in no_proxy:
if netloc.endswith(host) or netloc.split(":")[0].endswith(host):
if parsed.netloc.endswith(host) or parsed.netloc.split(":")[0].endswith(host):
# The URL does match something in no_proxy, so we don't want
# to apply the proxies on this URL.
return True
with set_environ("no_proxy", no_proxy_arg):
return bool(proxy_bypass(netloc))
return bool(proxy_bypass(parsed.netloc))
def get_environ_proxies(url: str, no_proxy: typing.Optional[bool] = None) -> dict:
setup.py
+3 -2
View File
@@ -37,8 +37,9 @@ class PyTest(Command):
def run(self):
import pytest
errno = pytest.main(["-n", "auto"])
# errno = pytest.main([])
# errno = pytest.main(["-n", "auto"])
errno = pytest.main([])
sys.exit(errno)
tests/test_help.py
+2 -2
View File
@@ -29,11 +29,11 @@ def test_idna_without_version_attribute(mocker):
"""Older versions of IDNA don't provide a __version__ attribute, verify
that if we have such a package, we don't blow up.
"""
mocker.patch("requests.help.idna", new=None)
mocker.patch("requests3.help.idna", new=None)
assert info()["idna"] == {"version": ""}
def test_idna_with_version_attribute(mocker):
"""Verify we're actually setting idna version when it should be available."""
mocker.patch("requests.help.idna", new=VersionedPackage("2.6"))
mocker.patch("requests3.help.idna", new=VersionedPackage("2.6"))
assert info()["idna"] == {"version": "2.6"}
tests/test_lowlevel.py
-232
View File
@@ -1,232 +0,0 @@
# -*- coding: utf-8 -*-
import pytest
import threading
import requests3 as requests
from tests.testserver.server import Server, consume_socket_content
from .utils import override_environ
def test_chunked_upload():
"""can safely send generators"""
close_server = threading.Event()
server = Server.basic_response_server(wait_to_close_event=close_server)
data = iter([b"a", b"b", b"c"])
with server as (host, port):
url = "http://{}:{}/".format(host, port)
r = requests.post(url, data=data, stream=True)
close_server.set() # release server block
assert r.status_code == 200
assert r.request.headers["Transfer-Encoding"] == "chunked"
def test_incorrect_content_length():
"""Test ConnectionError raised for incomplete responses"""
close_server = threading.Event()
server = Server.text_response_server(
"HTTP/1.1 200 OK\r\n" + "Content-Length: 50\r\n\r\n" + "Hello World."
)
with server as (host, port):
url = "http://{0}:{1}/".format(host, port)
r = requests.Request("GET", url).prepare()
s = requests.Session()
with pytest.raises(requests.exceptions.ConnectionError) as e:
resp = s.send(r)
assert "12 bytes read, 38 more expected" in str(e)
close_server.set() # release server block
def test_digestauth_401_count_reset_on_redirect():
"""Ensure we correctly reset num_401_calls after a successful digest auth,
followed by a 302 redirect to another digest auth prompt.
See https://github.com/requests/requests/issues/1979.
"""
text_401 = (
b"HTTP/1.1 401 UNAUTHORIZED\r\n"
b"Content-Length: 0\r\n"
b'WWW-Authenticate: Digest nonce="6bf5d6e4da1ce66918800195d6b9130d"'
b', opaque="372825293d1c26955496c80ed6426e9e", '
b'realm="me@kennethreitz.com", qop=auth\r\n\r\n'
)
text_302 = b"HTTP/1.1 302 FOUND\r\n" b"Content-Length: 0\r\n" b"Location: /\r\n\r\n"
text_200 = b"HTTP/1.1 200 OK\r\n" b"Content-Length: 0\r\n\r\n"
expected_digest = (
b'Authorization: Digest username="user", '
b'realm="me@kennethreitz.com", '
b'nonce="6bf5d6e4da1ce66918800195d6b9130d", uri="/"'
)
auth = requests.auth.HTTPDigestAuth("user", "pass")
def digest_response_handler(sock):
# Respond to initial GET with a challenge.
request_content = consume_socket_content(sock, timeout=0.5)
assert request_content.startswith(b"GET / HTTP/1.1")
sock.send(text_401)
# Verify we receive an Authorization header in response, then redirect.
request_content = consume_socket_content(sock, timeout=0.5)
assert expected_digest in request_content
sock.send(text_302)
# Verify Authorization isn't sent to the redirected host,
# then send another challenge.
request_content = consume_socket_content(sock, timeout=0.5)
assert b"Authorization:" not in request_content
sock.send(text_401)
# Verify Authorization is sent correctly again, and return 200 OK.
request_content = consume_socket_content(sock, timeout=0.5)
assert expected_digest in request_content
sock.send(text_200)
return request_content
close_server = threading.Event()
server = Server(digest_response_handler, wait_to_close_event=close_server)
with server as (host, port):
url = "http://{}:{}/".format(host, port)
r = requests.get(url, auth=auth)
# Verify server succeeded in authenticating.
assert r.status_code == 200
# Verify Authorization was sent in final request.
assert "Authorization" in r.request.headers
assert r.request.headers["Authorization"].startswith("Digest ")
# Verify redirect happened as we expected.
assert r.history[0].status_code == 302
close_server.set()
def test_digestauth_401_only_sent_once():
"""Ensure we correctly respond to a 401 challenge once, and then
stop responding if challenged again.
"""
text_401 = (
b"HTTP/1.1 401 UNAUTHORIZED\r\n"
b"Content-Length: 0\r\n"
b'WWW-Authenticate: Digest nonce="6bf5d6e4da1ce66918800195d6b9130d"'
b', opaque="372825293d1c26955496c80ed6426e9e", '
b'realm="me@kennethreitz.com", qop=auth\r\n\r\n'
)
expected_digest = (
b'Authorization: Digest username="user", '
b'realm="me@kennethreitz.com", '
b'nonce="6bf5d6e4da1ce66918800195d6b9130d", uri="/"'
)
auth = requests.auth.HTTPDigestAuth("user", "pass")
def digest_failed_response_handler(sock):
# Respond to initial GET with a challenge.
request_content = consume_socket_content(sock, timeout=0.5)
assert request_content.startswith(b"GET / HTTP/1.1")
sock.send(text_401)
# Verify we receive an Authorization header in response, then
# challenge again.
request_content = consume_socket_content(sock, timeout=0.5)
assert expected_digest in request_content
sock.send(text_401)
# Verify the client didn't respond to second challenge.
request_content = consume_socket_content(sock, timeout=0.5)
assert request_content == b""
return request_content
close_server = threading.Event()
server = Server(digest_failed_response_handler, wait_to_close_event=close_server)
with server as (host, port):
url = "http://{}:{}/".format(host, port)
r = requests.get(url, auth=auth)
# Verify server didn't authenticate us.
assert r.status_code == 401
assert r.history[0].status_code == 401
close_server.set()
def test_digestauth_only_on_4xx():
"""Ensure we only send digestauth on 4xx challenges.
See https://github.com/requests/requests/issues/3772.
"""
text_200_chal = (
b"HTTP/1.1 200 OK\r\n"
b"Content-Length: 0\r\n"
b'WWW-Authenticate: Digest nonce="6bf5d6e4da1ce66918800195d6b9130d"'
b', opaque="372825293d1c26955496c80ed6426e9e", '
b'realm="me@kennethreitz.com", qop=auth\r\n\r\n'
)
auth = requests.auth.HTTPDigestAuth("user", "pass")
def digest_response_handler(sock):
# Respond to GET with a 200 containing www-authenticate header.
request_content = consume_socket_content(sock, timeout=0.5)
assert request_content.startswith(b"GET / HTTP/1.1")
sock.send(text_200_chal)
# Verify the client didn't respond with auth.
request_content = consume_socket_content(sock, timeout=0.5)
assert request_content == b""
return request_content
close_server = threading.Event()
server = Server(digest_response_handler, wait_to_close_event=close_server)
with server as (host, port):
url = "http://{}:{}/".format(host, port)
r = requests.get(url, auth=auth)
# Verify server didn't receive auth from us.
assert r.status_code == 200
assert len(r.history) == 0
close_server.set()
_schemes_by_var_prefix = [
("http", ["http"]),
("https", ["https"]),
("all", ["http", "https"]),
]
_proxy_combos = []
for prefix, schemes in _schemes_by_var_prefix:
for scheme in schemes:
_proxy_combos.append(("{0}_proxy".format(prefix), scheme))
_proxy_combos += [(var.upper(), scheme) for var, scheme in _proxy_combos]
@pytest.mark.parametrize("var,scheme", _proxy_combos)
def test_use_proxy_from_environment(httpbin, var, scheme):
url = "{0}://httpbin.org".format(scheme)
fake_proxy = Server() # do nothing with the requests; just close the socket
with fake_proxy as (host, port):
proxy_url = "socks5://{}:{}".format(host, port)
kwargs = {var: proxy_url}
with override_environ(**kwargs):
# fake proxy's lack of response will cause a ConnectionError
with pytest.raises(requests.exceptions.ConnectionError):
requests.get(url)
# the fake proxy received a request
assert len(fake_proxy.handler_results) == 1
# it had actual content (not checking for SOCKS protocol for now)
assert len(fake_proxy.handler_results[0]) > 0
def test_redirect_rfc1808_to_non_ascii_location():
path = u"š"
expected_path = b"%C5%A1"
redirect_request = [] # stores the second request to the server
def redirect_resp_handler(sock):
consume_socket_content(sock, timeout=0.5)
location = u"//{}:{}/{}".format(host, port, path)
sock.send(
b"HTTP/1.1 301 Moved Permanently\r\n"
b"Content-Length: 0\r\n"
b"Location: " + location.encode("utf8") + b"\r\n"
b"\r\n"
)
redirect_request.append(consume_socket_content(sock, timeout=0.5))
sock.send(b"HTTP/1.1 200 OK\r\n\r\n")
close_server = threading.Event()
server = Server(redirect_resp_handler, wait_to_close_event=close_server)
with server as (host, port):
url = u"http://{}:{}".format(host, port)
r = requests.get(url=url, allow_redirects=True)
assert r.status_code == 200
assert len(r.history) == 1
assert r.history[0].status_code == 301
assert redirect_request[0].startswith(b"GET /" + expected_path + b" HTTP/1.1")
assert r.url == u"{0}/{1}".format(url, expected_path.decode("ascii"))
close_server.set()
tests/test_requests.py
-3026
View File
File diff suppressed because it is too large Load Diff
tests/test_testserver.py
-155
View File
@@ -1,155 +0,0 @@
# -*- coding: utf-8 -*-
import threading
import socket
import time
import pytest
import requests3 as requests
from tests.testserver.server import Server
class TestTestServer:
def test_basic(self):
"""messages are sent and received properly"""
question = b"success?"
answer = b"yeah, success"
def handler(sock):
text = sock.recv(1000)
assert text == question
sock.sendall(answer)
with Server(handler) as (host, port):
sock = socket.socket()
sock.connect((host, port))
sock.sendall(question)
text = sock.recv(1000)
assert text == answer
sock.close()
def test_server_closes(self):
"""the server closes when leaving the context manager"""
with Server.basic_response_server() as (host, port):
sock = socket.socket()
sock.connect((host, port))
sock.close()
with pytest.raises(socket.error):
new_sock = socket.socket()
new_sock.connect((host, port))
def test_text_response(self):
"""the text_response_server sends the given text"""
server = Server.text_response_server(
"HTTP/1.1 200 OK\r\n" + "Content-Length: 6\r\n" + "\r\nroflol"
)
with server as (host, port):
r = requests.get('http://{0}:{1}'.format(host, port))
assert r.status_code == 200
assert r.text == u'roflol'
assert r.headers['Content-Length'] == '6'
def test_basic_response(self):
"""the basic response server returns an empty http response"""
with Server.basic_response_server() as (host, port):
r = requests.get('http://{}:{}'.format(host, port))
assert r.status_code == 200
assert r.text == u''
assert r.headers['Content-Length'] == '0'
def test_basic_waiting_server(self):
"""the server waits for the block_server event to be set before closing"""
block_server = threading.Event()
with Server.basic_response_server(
wait_to_close_event=block_server
) as (
host, port
):
sock = socket.socket()
sock.connect((host, port))
sock.sendall(b'send something')
time.sleep(2.5)
sock.sendall(b'still alive')
block_server.set() # release server block
def test_multiple_requests(self):
"""multiple requests can be served"""
requests_to_handle = 5
server = Server.basic_response_server(
requests_to_handle=requests_to_handle
)
with server as (host, port):
server_url = 'http://{}:{}'.format(host, port)
for _ in range(requests_to_handle):
r = requests.get(server_url)
assert r.status_code == 200
# the (n+1)th request fails
with pytest.raises(requests.exceptions.ConnectionError):
r = requests.get(server_url)
@pytest.mark.skip(
reason="this fails non-deterministically under pytest-xdist"
)
def test_request_recovery(self):
"""can check the requests content"""
# TODO: figure out why this sometimes fails when using pytest-xdist.
server = Server.basic_response_server(requests_to_handle=2)
first_request = b'put your hands up in the air'
second_request = b'put your hand down in the floor'
with server as address:
sock1 = socket.socket()
sock2 = socket.socket()
sock1.connect(address)
sock1.sendall(first_request)
sock1.close()
sock2.connect(address)
sock2.sendall(second_request)
sock2.close()
assert server.handler_results[0] == first_request
assert server.handler_results[1] == second_request
def test_requests_after_timeout_are_not_received(self):
"""the basic response handler times out when receiving requests"""
server = Server.basic_response_server(request_timeout=1)
with server as address:
sock = socket.socket()
sock.connect(address)
time.sleep(1.5)
sock.sendall(b'hehehe, not received')
sock.close()
assert server.handler_results[0] == b''
def test_request_recovery_with_bigger_timeout(self):
"""a biggest timeout can be specified"""
server = Server.basic_response_server(request_timeout=3)
data = b'bananadine'
with server as address:
sock = socket.socket()
sock.connect(address)
time.sleep(1.5)
sock.sendall(data)
sock.close()
assert server.handler_results[0] == data
def test_server_finishes_on_error(self):
"""the server thread exits even if an exception exits the context manager"""
server = Server.basic_response_server()
with pytest.raises(Exception):
with server:
raise Exception()
assert len(server.handler_results) == 0
# if the server thread fails to finish, the test suite will hang
# and get killed by the jenkins timeout.
def test_server_finishes_when_no_connections(self):
"""the server thread exits even if there are no connections"""
server = Server.basic_response_server()
with server:
pass
assert len(server.handler_results) == 0
# if the server thread fails to finish, the test suite will hang
# and get killed by the jenkins timeout.
tests/test_utils.py
+1
View File
@@ -35,6 +35,7 @@ from requests3.http_utils import (
urldefragauth,
add_dict_to_cookiejar,
set_environ,
extract_zipped_paths
)
from requests3._internal_utils import unicode_is_ascii