$ make ci 
pytest tests --junitxml=report.xml 
======================================================================================================= test session starts ======================================================================================================= 
platform linux -- Python 3.6.8, pytest-3.10.1, py-1.10.0, pluggy-0.13.1 
rootdir: /home/work6/project/requests, inifile: pytest.ini 
plugins: mock-2.0.0, httpbin-1.0.0, cov-2.9.0 
collected 552 items
 
tests/test_help.py ...  [  0%] 
tests/test_hooks.py ... [  1%] 
tests/test_lowlevel.py ...............  [  3%] 
tests/test_packages.py ...  [  4%] 
tests/test_requests.py .................................................................................................................................................................................................... [ 39%] 
127.0.0.1 - - [10/Aug/2021 08:41:53] "GET /stream/4 HTTP/1.1" 200 756 
.127.0.0.1 - - [10/Aug/2021 08:41:53] "GET /stream/4 HTTP/1.1" 500 59 
---------------------------------------- 
Exception happened during processing of request from ('127.0.0.1', 46048) 
Traceback (most recent call last): 
  File "/usr/lib64/python3.6/wsgiref/handlers.py", line 138, in run 
 self.finish_response() 
x.........................................................................................  [ 56%] 
tests/test_structures.py .................... [ 59%] 
tests/test_testserver.py ......s.... [ 61%] 
tests/test_utils.py ..s................................................................................................................................................................................................ssss [ 98%] 
ssssss..... [100%] 
 
----------------------------------------------------------------------------------- generated xml file: /home/work6/project/requests/report.xml ----------------------------------------------------------------------------------- 
======================================================================================= 539 passed, 12 skipped, 1 xfailed in 64.16 seconds ======================================================================================== 

$ make coverage 
----------- coverage: platform linux, python 3.6.8-final-0 ----------- 
Name  StmtsMiss  Cover 
------------------------------------------------- 
requests/__init__.py 71  71  0% 
requests/__version__.py 10  10  0% 
requests/_internal_utils.py165 69% 
requests/adapters.py222  67 70% 
requests/api.py20  13 35% 
requests/auth.py 174  54 69% 
requests/certs.py  44  0% 
requests/compat.py47  47  0% 
requests/cookies.py 238 115 52% 
requests/exceptions.py  35  29 17% 
requests/help.py  63  19 70% 
requests/hooks.py 154 73% 
requests/models.py  455 119 74% 
requests/packages.py 16  16  0% 
requests/sessions.py283  67 76% 
requests/status_codes.py15  15  0% 
requests/structures.py  40  19 52% 
requests/utils.py465 170 63% 
------------------------------------------------- 
TOTAL  2189 844 61% 
Coverage XML written to file coverage.xml 

import idna 
 
def _implementation(): 
 ... 
  
def info(): 
 ... 
 system_ssl = ssl.OPENSSL_VERSION_NUMBER 
 system_ssl_info = { 
  'version': '%x' % system_ssl if system_ssl is not None else '' 
 } 
 idna_info = { 
  'version': getattr(idna, '__version__', ''), 
 } 
 ... 
 return { 
  'platform': platform_info, 
  'implementation': implementation_info, 
  'system_ssl': system_ssl_info, 
  'using_pyopenssl': pyopenssl is not None, 
  'pyOpenSSL': pyopenssl_info, 
  'urllib3': urllib3_info, 
  'chardet': chardet_info, 
  'cryptography': cryptography_info, 
  'idna': idna_info, 
  'requests': { 
'version': requests_version, 
  }, 
 } 

from requests.help import info 
 
def test_system_ssl(): 
 """Verify we're actually setting system_ssl when it should be available.""" 
 assert info()['system_ssl']['version'] != '' 
 
class VersionedPackage(object): 
 def __init__(self, version): 
  self.__version__ = version 
 
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) 
 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')) 
 assert info()['idna'] == {'version': '2.6'} 

# 模擬空實(shí)現(xiàn) 
mocker.patch('requests.help.idna', new=None) 
# 模擬版本2.6 
mocker.patch('requests.help.idna', new=VersionedPackage('2.6')) 

for package in ('urllib3', 'idna', 'chardet'): 
 locals()[package] = __import__(package) 
 # This traversal is apparently necessary such that the identities are 
 # preserved (requests.packages.urllib3.* is urllib3.*) 
 for mod in list(sys.modules): 
  if mod == package or mod.startswith(package + '.'): 
sys.modules['requests.packages.' + mod] = sys.modules[mod] 

from requests import hooks 
 
def hook(value): 
 return value[1:] 
 
@pytest.mark.parametrize( 
 'hooks_list, result', ( 
  (hook, 'ata'), 
  ([hook, lambda x: None, hook], 'ta'), 
 ) 
) 
def test_hooks(hooks_list, result): 
 assert hooks.dispatch_hook('response', {'response': hooks_list}, 'Data') == result 
 
def test_default_hooks(): 
 assert hooks.default_hooks() == {'response': []} 

def dispatch_hook(key, hooks, hook_data, **kwargs): 
 """Dispatches a hook dictionary on a given piece of data.""" 
 hooks = hooks or {} 
 hooks = hooks.get(key) 
 if hooks: 
  # 判斷鉤子函數(shù) 
  if hasattr(hooks, '__call__'): 
hooks = [hooks] 
  for hook in hooks: 
_hook_data = hook(hook_data, **kwargs) 
if _hook_data is not None: 
 hook_data = _hook_data 
 return hook_data 

class TestLookupDict: 
 
 @pytest.fixture(autouse=True) 
 def setup(self): 
  """LookupDict instance with "bad_gateway" attribute.""" 
  self.lookup_dict = LookupDict('test') 
  self.lookup_dict.bad_gateway = 502 
 
 def test_repr(self): 
  assert repr(self.lookup_dict) == "<lookup 'test'>" 
 
 get_item_parameters = pytest.mark.parametrize( 
  'key, value', ( 
('bad_gateway', 502), 
('not_a_key', None) 
  ) 
 ) 
 
 @get_item_parameters 
 def test_getitem(self, key, value): 
  assert self.lookup_dict[key] == value 
 
 @get_item_parameters 
 def test_get(self, key, value): 
  assert self.lookup_dict.get(key) == value 

class LookupDict(dict): 
 ... 
 def __getitem__(self, key): 
  # We allow fall-through here, so values default to None 
  return self.__dict__.get(key, None) 
 
 def get(self, key, default=None): 
  return self.__dict__.get(key, default) 

class TestCaseInsensitiveDict:
# 類測(cè)試
def test_repr(self):
assert repr(self.case_insensitive_dict) == "{'Accept': 'application/json'}"
def test_copy(self):
copy = self.case_insensitive_dict.copy()
assert copy is not self.case_insensitive_dict
assert copy == self.case_insensitive_dict
class TestCaseInsensitiveDict:
# 使用方法測(cè)試
def test_delitem(self):
cid = CaseInsensitiveDict()
cid['Spam'] = 'someval'
del cid['sPam']
assert 'spam' not in cid
assert len(cid) == 0
def test_contains(self):
cid = CaseInsensitiveDict()
cid['Spam'] = 'someval'
assert 'Spam' in cid
assert 'spam' in cid
assert 'SPAM' in cid
assert 'sPam' in cid
assert 'notspam' not in cid

import contextlib
import os
@contextlib.contextmanager
def override_environ(**kwargs):
save_env = dict(os.environ)
for key, value in kwargs.items():
if value is None:
del os.environ[key]
else:
os.environ[key] = value
try:
yield
finally:
os.environ.clear()
os.environ.update(save_env)

# test_requests.py
kwargs = {
var: proxy
}
# 模擬控制proxy環(huán)境變量
with override_environ(**kwargs):
proxies = session.rebuild_proxies(prep, {})
def rebuild_proxies(self, prepared_request, proxies):
bypass_proxy = should_bypass_proxies(url, no_proxy=no_proxy)
def should_bypass_proxies(url, no_proxy):
...
get_proxy = lambda k: os.environ.get(k) or os.environ.get(k.upper())
...

# 對(duì)象轉(zhuǎn)列表
def to_key_val_list(value):
if value is None:
return None
if isinstance(value, (str, bytes, bool, int)):
raise ValueError('cannot encode objects that are not 2-tuples')
if isinstance(value, Mapping):
value = value.items()
return list(value)

class TestToKeyValList:
@pytest.mark.parametrize(
'value, expected', (
([('key', 'val')], [('key', 'val')]),
((('key', 'val'), ), [('key', 'val')]),
({'key': 'val'}, [('key', 'val')]),
(None, None)
))
def test_valid(self, value, expected):
assert to_key_val_list(value) == expected
def test_invalid(self):
with pytest.raises(ValueError):
to_key_val_list('string')

class TestSuperLen:
@pytest.mark.parametrize(
'stream, value', (
(StringIO.StringIO, 'Test'),
(BytesIO, b'Test'),
pytest.param(cStringIO, 'Test',
marks=pytest.mark.skipif('cStringIO is None')),
))
def test_io_streams(self, stream, value):
"""Ensures that we properly deal with different kinds of IO streams."""
assert super_len(stream()) == 0
assert super_len(stream(value)) == 4
def test_super_len_correctly_calculates_len_of_partially_read_file(self):
"""Ensure that we handle partially consumed file like objects."""
s = StringIO.StringIO()
s.write('foobarbogus')
assert super_len(s) == 0
@pytest.mark.parametrize(
'mode, warnings_num', (
('r', 1),
('rb', 0),
))
def test_file(self, tmpdir, mode, warnings_num, recwarn):
file_obj = tmpdir.join('test.txt')
file_obj.write('Test')
with file_obj.open(mode) as fd:
assert super_len(fd) == 4
assert len(recwarn) == warnings_num
def test_super_len_with_tell(self):
foo = StringIO.StringIO('12345')
assert super_len(foo) == 5
foo.read(2)
assert super_len(foo) == 3
def test_super_len_with_fileno(self):
with open(__file__, 'rb') as f:
length = super_len(f)
file_data = f.read()
assert length == len(file_data)

TARPIT = 'http://10.255.255.1'
class TestTimeout:
def test_stream_timeout(self, httpbin):
try:
requests.get(httpbin('delay/10'), timeout=2.0)
except requests.exceptions.Timeout as e:
assert 'Read timed out' in e.args[0].args[0]
@pytest.mark.parametrize(
'timeout', (
(0.1, None),
Urllib3Timeout(connect=0.1, read=None)
))
def test_connect_timeout(self, timeout):
try:
requests.get(TARPIT, timeout=timeout)
pytest.fail('The connect() request should time out.')
except ConnectTimeout as e:
assert isinstance(e, ConnectionError)
assert isinstance(e, Timeout)

class TestRequests:
def test_basic_building(self):
req = requests.Request()
req.url = 'http://kennethreitz.org/'
req.data = {'life': '42'}
pr = req.prepare()
assert pr.url == req.url
assert pr.body == 'life=42'
def test_path_is_not_double_encoded(self):
request = requests.Request('GET', "http://0.0.0.0/get/test case").prepare()
assert request.path_url == '/get/test%20case
...
def test_HTTP_200_OK_GET_ALTERNATIVE(self, httpbin):
r = requests.Request('GET', httpbin('get'))
s = requests.Session()
s.proxies = getproxies()
r = s.send(r.prepare())
assert r.status_code == 200
ef test_set_cookie_on_301(self, httpbin):
s = requests.session()
url = httpbin('cookies/set?foo=bar')
s.get(url)
assert s.cookies['foo'] == 'bar'

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_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://{}:{}'.format(host, port))
assert r.status_code == 200
assert r.text == u'roflol'
assert r.headers['Content-Length'] == '6'

def test_fragment_not_sent_with_request():
"""Verify that the fragment portion of a URI isn't sent to the server."""
def response_handler(sock):
req = consume_socket_content(sock, timeout=0.5)
sock.send(
b'HTTP/1.1 200 OK\r\n'
b'Content-Length: '+bytes(len(req))+b'\r\n'
b'\r\n'+req
)
close_server = threading.Event()
server = Server(response_handler, wait_to_close_event=close_server)
with server as (host, port):
url = 'http://{}:{}/path/to/thing/#view=edit&token=hunter2'.format(host, port)
r = requests.get(url)
raw_request = r.content
assert r.status_code == 200
headers, body = raw_request.split(b'\r\n\r\n', 1)
status_line, headers = headers.split(b'\r\n', 1)
assert status_line == b'GET /path/to/thing/ HTTP/1.1'
for frag in (b'view', b'edit', b'token', b'hunter2'):
assert frag not in headers
assert frag not in body
close_server.set()