Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/urllib3/_base_connection.py: 26%

1from __future__ import annotations 

2 

3import typing 

4 

5from .util.connection import _TYPE_SOCKET_OPTIONS 

6from .util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT 

7from .util.url import Url 

8 

9_TYPE_BODY = typing.Union[bytes, typing.IO[typing.Any], typing.Iterable[bytes], str] 

10 

11 

12class ProxyConfig(typing.NamedTuple): 

13 ssl_context: ssl.SSLContext | None 

14 use_forwarding_for_https: bool 

15 assert_hostname: None | str | Literal[False] 

16 assert_fingerprint: str | None 

17 

18 

19class _ResponseOptions(typing.NamedTuple): 

20 # TODO: Remove this in favor of a better 

21 # HTTP request/response lifecycle tracking. 

22 request_method: str 

23 request_url: str 

24 preload_content: bool 

25 decode_content: bool 

26 enforce_content_length: bool 

27 

28 

29if typing.TYPE_CHECKING: 

30 import ssl 

31 

32 from typing_extensions import Literal, Protocol 

33 

34 from .response import BaseHTTPResponse 

35 

36 class BaseHTTPConnection(Protocol): 

37 default_port: typing.ClassVar[int] 

38 default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] 

39 

40 host: str 

41 port: int 

42 timeout: None | ( 

43 float 

44 ) # Instance doesn't store _DEFAULT_TIMEOUT, must be resolved. 

45 blocksize: int 

46 source_address: tuple[str, int] | None 

47 socket_options: _TYPE_SOCKET_OPTIONS | None 

48 

49 proxy: Url | None 

50 proxy_config: ProxyConfig | None 

51 

52 is_verified: bool 

53 proxy_is_verified: bool | None 

54 

55 def __init__( 

56 self, 

57 host: str, 

58 port: int | None = None, 

59 *, 

60 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

61 source_address: tuple[str, int] | None = None, 

62 blocksize: int = 8192, 

63 socket_options: _TYPE_SOCKET_OPTIONS | None = ..., 

64 proxy: Url | None = None, 

65 proxy_config: ProxyConfig | None = None, 

66 ) -> None: 

67 ... 

68 

69 def set_tunnel( 

70 self, 

71 host: str, 

72 port: int | None = None, 

73 headers: typing.Mapping[str, str] | None = None, 

74 scheme: str = "http", 

75 ) -> None: 

76 ... 

77 

78 def connect(self) -> None: 

79 ... 

80 

81 def request( 

82 self, 

83 method: str, 

84 url: str, 

85 body: _TYPE_BODY | None = None, 

86 headers: typing.Mapping[str, str] | None = None, 

87 # We know *at least* botocore is depending on the order of the 

88 # first 3 parameters so to be safe we only mark the later ones 

89 # as keyword-only to ensure we have space to extend. 

90 *, 

91 chunked: bool = False, 

92 preload_content: bool = True, 

93 decode_content: bool = True, 

94 enforce_content_length: bool = True, 

95 ) -> None: 

96 ... 

97 

98 def getresponse(self) -> BaseHTTPResponse: 

99 ... 

100 

101 def close(self) -> None: 

102 ... 

103 

104 @property 

105 def is_closed(self) -> bool: 

106 """Whether the connection either is brand new or has been previously closed. 

107 If this property is True then both ``is_connected`` and ``has_connected_to_proxy`` 

108 properties must be False. 

109 """ 

110 

111 @property 

112 def is_connected(self) -> bool: 

113 """Whether the connection is actively connected to any origin (proxy or target)""" 

114 

115 @property 

116 def has_connected_to_proxy(self) -> bool: 

117 """Whether the connection has successfully connected to its proxy. 

118 This returns False if no proxy is in use. Used to determine whether 

119 errors are coming from the proxy layer or from tunnelling to the target origin. 

120 """ 

121 

122 class BaseHTTPSConnection(BaseHTTPConnection, Protocol): 

123 default_port: typing.ClassVar[int] 

124 default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] 

125 

126 # Certificate verification methods 

127 cert_reqs: int | str | None 

128 assert_hostname: None | str | Literal[False] 

129 assert_fingerprint: str | None 

130 ssl_context: ssl.SSLContext | None 

131 

132 # Trusted CAs 

133 ca_certs: str | None 

134 ca_cert_dir: str | None 

135 ca_cert_data: None | str | bytes 

136 

137 # TLS version 

138 ssl_minimum_version: int | None 

139 ssl_maximum_version: int | None 

140 ssl_version: int | str | None # Deprecated 

141 

142 # Client certificates 

143 cert_file: str | None 

144 key_file: str | None 

145 key_password: str | None 

146 

147 def __init__( 

148 self, 

149 host: str, 

150 port: int | None = None, 

151 *, 

152 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

153 source_address: tuple[str, int] | None = None, 

154 blocksize: int = 8192, 

155 socket_options: _TYPE_SOCKET_OPTIONS | None = ..., 

156 proxy: Url | None = None, 

157 proxy_config: ProxyConfig | None = None, 

158 cert_reqs: int | str | None = None, 

159 assert_hostname: None | str | Literal[False] = None, 

160 assert_fingerprint: str | None = None, 

161 server_hostname: str | None = None, 

162 ssl_context: ssl.SSLContext | None = None, 

163 ca_certs: str | None = None, 

164 ca_cert_dir: str | None = None, 

165 ca_cert_data: None | str | bytes = None, 

166 ssl_minimum_version: int | None = None, 

167 ssl_maximum_version: int | None = None, 

168 ssl_version: int | str | None = None, # Deprecated 

169 cert_file: str | None = None, 

170 key_file: str | None = None, 

171 key_password: str | None = None, 

172 ) -> None: 

173 ...