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 from typing import Literal, Protocol 

32 

33 from .response import BaseHTTPResponse 

34 

35 class BaseHTTPConnection(Protocol): 

36 default_port: typing.ClassVar[int] 

37 default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] 

38 

39 host: str 

40 port: int 

41 timeout: None | ( 

42 float 

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

44 blocksize: int 

45 source_address: tuple[str, int] | None 

46 socket_options: _TYPE_SOCKET_OPTIONS | None 

47 

48 proxy: Url | None 

49 proxy_config: ProxyConfig | None 

50 

51 is_verified: bool 

52 proxy_is_verified: bool | None 

53 

54 def __init__( 

55 self, 

56 host: str, 

57 port: int | None = None, 

58 *, 

59 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

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

61 blocksize: int = 8192, 

62 socket_options: _TYPE_SOCKET_OPTIONS | None = ..., 

63 proxy: Url | None = None, 

64 proxy_config: ProxyConfig | None = None, 

65 ) -> None: 

66 ... 

67 

68 def set_tunnel( 

69 self, 

70 host: str, 

71 port: int | None = None, 

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

73 scheme: str = "http", 

74 ) -> None: 

75 ... 

76 

77 def connect(self) -> None: 

78 ... 

79 

80 def request( 

81 self, 

82 method: str, 

83 url: str, 

84 body: _TYPE_BODY | None = None, 

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

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

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

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

89 *, 

90 chunked: bool = False, 

91 preload_content: bool = True, 

92 decode_content: bool = True, 

93 enforce_content_length: bool = True, 

94 ) -> None: 

95 ... 

96 

97 def getresponse(self) -> BaseHTTPResponse: 

98 ... 

99 

100 def close(self) -> None: 

101 ... 

102 

103 @property 

104 def is_closed(self) -> bool: 

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

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

107 properties must be False. 

108 """ 

109 

110 @property 

111 def is_connected(self) -> bool: 

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

113 

114 @property 

115 def has_connected_to_proxy(self) -> bool: 

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

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

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

119 """ 

120 

121 class BaseHTTPSConnection(BaseHTTPConnection, Protocol): 

122 default_port: typing.ClassVar[int] 

123 default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] 

124 

125 # Certificate verification methods 

126 cert_reqs: int | str | None 

127 assert_hostname: None | str | Literal[False] 

128 assert_fingerprint: str | None 

129 ssl_context: ssl.SSLContext | None 

130 

131 # Trusted CAs 

132 ca_certs: str | None 

133 ca_cert_dir: str | None 

134 ca_cert_data: None | str | bytes 

135 

136 # TLS version 

137 ssl_minimum_version: int | None 

138 ssl_maximum_version: int | None 

139 ssl_version: int | str | None # Deprecated 

140 

141 # Client certificates 

142 cert_file: str | None 

143 key_file: str | None 

144 key_password: str | None 

145 

146 def __init__( 

147 self, 

148 host: str, 

149 port: int | None = None, 

150 *, 

151 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

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

153 blocksize: int = 16384, 

154 socket_options: _TYPE_SOCKET_OPTIONS | None = ..., 

155 proxy: Url | None = None, 

156 proxy_config: ProxyConfig | None = None, 

157 cert_reqs: int | str | None = None, 

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

159 assert_fingerprint: str | None = None, 

160 server_hostname: str | None = None, 

161 ssl_context: ssl.SSLContext | None = None, 

162 ca_certs: str | None = None, 

163 ca_cert_dir: str | None = None, 

164 ca_cert_data: None | str | bytes = None, 

165 ssl_minimum_version: int | None = None, 

166 ssl_maximum_version: int | None = None, 

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

168 cert_file: str | None = None, 

169 key_file: str | None = None, 

170 key_password: str | None = None, 

171 ) -> None: 

172 ...