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

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[ 

10 bytes, typing.IO[typing.Any], typing.Iterable[bytes | str], str 

11] 

12 

13 

14class ProxyConfig(typing.NamedTuple): 

15 ssl_context: ssl.SSLContext | None 

16 use_forwarding_for_https: bool 

17 assert_hostname: None | str | typing.Literal[False] 

18 assert_fingerprint: str | None 

19 

20 

21class _ResponseOptions(typing.NamedTuple): 

22 # TODO: Remove this in favor of a better 

23 # HTTP request/response lifecycle tracking. 

24 request_method: str 

25 request_url: str 

26 preload_content: bool 

27 decode_content: bool 

28 enforce_content_length: bool 

29 

30 

31if typing.TYPE_CHECKING: 

32 import ssl 

33 from typing import Protocol 

34 

35 from .response import BaseHTTPResponse 

36 

37 class BaseHTTPConnection(Protocol): 

38 default_port: typing.ClassVar[int] 

39 default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] 

40 

41 host: str 

42 port: int 

43 timeout: None | ( 

44 float 

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

46 blocksize: int 

47 source_address: tuple[str, int] | None 

48 socket_options: _TYPE_SOCKET_OPTIONS | None 

49 

50 proxy: Url | None 

51 proxy_config: ProxyConfig | None 

52 

53 is_verified: bool 

54 proxy_is_verified: bool | None 

55 

56 def __init__( 

57 self, 

58 host: str, 

59 port: int | None = None, 

60 *, 

61 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

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

63 blocksize: int = 8192, 

64 socket_options: _TYPE_SOCKET_OPTIONS | None = ..., 

65 proxy: Url | None = None, 

66 proxy_config: ProxyConfig | None = None, 

67 ) -> None: ... 

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 def connect(self) -> None: ... 

78 

79 def request( 

80 self, 

81 method: str, 

82 url: str, 

83 body: _TYPE_BODY | None = None, 

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

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

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

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

88 *, 

89 chunked: bool = False, 

90 preload_content: bool = True, 

91 decode_content: bool = True, 

92 enforce_content_length: bool = True, 

93 ) -> None: ... 

94 

95 def getresponse(self) -> BaseHTTPResponse: ... 

96 

97 def close(self) -> None: ... 

98 

99 @property 

100 def is_closed(self) -> bool: 

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

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

103 properties must be False. 

104 """ 

105 

106 @property 

107 def is_connected(self) -> bool: 

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

109 

110 @property 

111 def has_connected_to_proxy(self) -> bool: 

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

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

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

115 """ 

116 

117 class BaseHTTPSConnection(BaseHTTPConnection, Protocol): 

118 default_port: typing.ClassVar[int] 

119 default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] 

120 

121 # Certificate verification methods 

122 cert_reqs: int | str | None 

123 assert_hostname: None | str | typing.Literal[False] 

124 assert_fingerprint: str | None 

125 ssl_context: ssl.SSLContext | None 

126 

127 # Trusted CAs 

128 ca_certs: str | None 

129 ca_cert_dir: str | None 

130 ca_cert_data: None | str | bytes 

131 

132 # TLS version 

133 ssl_minimum_version: int | None 

134 ssl_maximum_version: int | None 

135 ssl_version: int | str | None # Deprecated 

136 

137 # Client certificates 

138 cert_file: str | None 

139 key_file: str | None 

140 key_password: str | None 

141 

142 def __init__( 

143 self, 

144 host: str, 

145 port: int | None = None, 

146 *, 

147 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

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

149 blocksize: int = 16384, 

150 socket_options: _TYPE_SOCKET_OPTIONS | None = ..., 

151 proxy: Url | None = None, 

152 proxy_config: ProxyConfig | None = None, 

153 cert_reqs: int | str | None = None, 

154 assert_hostname: None | str | typing.Literal[False] = None, 

155 assert_fingerprint: str | None = None, 

156 server_hostname: str | None = None, 

157 ssl_context: ssl.SSLContext | None = None, 

158 ca_certs: str | None = None, 

159 ca_cert_dir: str | None = None, 

160 ca_cert_data: None | str | bytes = None, 

161 ssl_minimum_version: int | None = None, 

162 ssl_maximum_version: int | None = None, 

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

164 cert_file: str | None = None, 

165 key_file: str | None = None, 

166 key_password: str | None = None, 

167 ) -> None: ...