Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_vendor/urllib3/util/request.py: 49%

1from __future__ import absolute_import 

2 

3from base64 import b64encode 

4 

5from ..exceptions import UnrewindableBodyError 

6from ..packages.six import b, integer_types 

7 

8# Pass as a value within ``headers`` to skip 

9# emitting some HTTP headers that are added automatically. 

10# The only headers that are supported are ``Accept-Encoding``, 

11# ``Host``, and ``User-Agent``. 

12SKIP_HEADER = "@@@SKIP_HEADER@@@" 

13SKIPPABLE_HEADERS = frozenset(["accept-encoding", "host", "user-agent"]) 

14 

15ACCEPT_ENCODING = "gzip,deflate" 

16 

17_FAILEDTELL = object() 

18 

19 

20def make_headers( 

21 keep_alive=None, 

22 accept_encoding=None, 

23 user_agent=None, 

24 basic_auth=None, 

25 proxy_basic_auth=None, 

26 disable_cache=None, 

27): 

28 """ 

29 Shortcuts for generating request headers. 

30 

31 :param keep_alive: 

32 If ``True``, adds 'connection: keep-alive' header. 

33 

34 :param accept_encoding: 

35 Can be a boolean, list, or string. 

36 ``True`` translates to 'gzip,deflate'. 

37 List will get joined by comma. 

38 String will be used as provided. 

39 

40 :param user_agent: 

41 String representing the user-agent you want, such as 

42 "python-urllib3/0.6" 

43 

44 :param basic_auth: 

45 Colon-separated username:password string for 'authorization: basic ...' 

46 auth header. 

47 

48 :param proxy_basic_auth: 

49 Colon-separated username:password string for 'proxy-authorization: basic ...' 

50 auth header. 

51 

52 :param disable_cache: 

53 If ``True``, adds 'cache-control: no-cache' header. 

54 

55 Example:: 

56 

57 >>> make_headers(keep_alive=True, user_agent="Batman/1.0") 

58 {'connection': 'keep-alive', 'user-agent': 'Batman/1.0'} 

59 >>> make_headers(accept_encoding=True) 

60 {'accept-encoding': 'gzip,deflate'} 

61 """ 

62 headers = {} 

63 if accept_encoding: 

64 if isinstance(accept_encoding, str): 

65 pass 

66 elif isinstance(accept_encoding, list): 

67 accept_encoding = ",".join(accept_encoding) 

68 else: 

69 accept_encoding = ACCEPT_ENCODING 

70 headers["accept-encoding"] = accept_encoding 

71 

72 if user_agent: 

73 headers["user-agent"] = user_agent 

74 

75 if keep_alive: 

76 headers["connection"] = "keep-alive" 

77 

78 if basic_auth: 

79 headers["authorization"] = "Basic " + b64encode(b(basic_auth)).decode("utf-8") 

80 

81 if proxy_basic_auth: 

82 headers["proxy-authorization"] = "Basic " + b64encode( 

83 b(proxy_basic_auth) 

84 ).decode("utf-8") 

85 

86 if disable_cache: 

87 headers["cache-control"] = "no-cache" 

88 

89 return headers 

90 

91 

92def set_file_position(body, pos): 

93 """ 

94 If a position is provided, move file to that point. 

95 Otherwise, we'll attempt to record a position for future use. 

96 """ 

97 if pos is not None: 

98 rewind_body(body, pos) 

99 elif getattr(body, "tell", None) is not None: 

100 try: 

101 pos = body.tell() 

102 except (IOError, OSError): 

103 # This differentiates from None, allowing us to catch 

104 # a failed `tell()` later when trying to rewind the body. 

105 pos = _FAILEDTELL 

106 

107 return pos 

108 

109 

110def rewind_body(body, body_pos): 

111 """ 

112 Attempt to rewind body to a certain position. 

113 Primarily used for request redirects and retries. 

114 

115 :param body: 

116 File-like object that supports seek. 

117 

118 :param int pos: 

119 Position to seek to in file. 

120 """ 

121 body_seek = getattr(body, "seek", None) 

122 if body_seek is not None and isinstance(body_pos, integer_types): 

123 try: 

124 body_seek(body_pos) 

125 except (IOError, OSError): 

126 raise UnrewindableBodyError( 

127 "An error occurred when rewinding request body for redirect/retry." 

128 ) 

129 elif body_pos is _FAILEDTELL: 

130 raise UnrewindableBodyError( 

131 "Unable to record file position for rewinding " 

132 "request body during a redirect/retry." 

133 ) 

134 else: 

135 raise ValueError( 

136 "body_pos must be of type integer, instead it was %s." % type(body_pos) 

137 )