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[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 | typing.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 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 def set_tunnel( 

68 self, 

69 host: str, 

70 port: int | None = None, 

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

72 scheme: str = "http", 

73 ) -> None: ... 

74 

75 def connect(self) -> None: ... 

76 

77 def request( 

78 self, 

79 method: str, 

80 url: str, 

81 body: _TYPE_BODY | None = None, 

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

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

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

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

86 *, 

87 chunked: bool = False, 

88 preload_content: bool = True, 

89 decode_content: bool = True, 

90 enforce_content_length: bool = True, 

91 ) -> None: ... 

92 

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

94 

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

96 

97 @property 

98 def is_closed(self) -> bool: 

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

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

101 properties must be False. 

102 """ 

103 

104 @property 

105 def is_connected(self) -> bool: 

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

107 

108 @property 

109 def has_connected_to_proxy(self) -> bool: 

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

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

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

113 """ 

114 

115 class BaseHTTPSConnection(BaseHTTPConnection, Protocol): 

116 default_port: typing.ClassVar[int] 

117 default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS] 

118 

119 # Certificate verification methods 

120 cert_reqs: int | str | None 

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

122 assert_fingerprint: str | None 

123 ssl_context: ssl.SSLContext | None 

124 

125 # Trusted CAs 

126 ca_certs: str | None 

127 ca_cert_dir: str | None 

128 ca_cert_data: None | str | bytes 

129 

130 # TLS version 

131 ssl_minimum_version: int | None 

132 ssl_maximum_version: int | None 

133 ssl_version: int | str | None # Deprecated 

134 

135 # Client certificates 

136 cert_file: str | None 

137 key_file: str | None 

138 key_password: str | None 

139 

140 def __init__( 

141 self, 

142 host: str, 

143 port: int | None = None, 

144 *, 

145 timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, 

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

147 blocksize: int = 16384, 

148 socket_options: _TYPE_SOCKET_OPTIONS | None = ..., 

149 proxy: Url | None = None, 

150 proxy_config: ProxyConfig | None = None, 

151 cert_reqs: int | str | None = None, 

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

153 assert_fingerprint: str | None = None, 

154 server_hostname: str | None = None, 

155 ssl_context: ssl.SSLContext | None = None, 

156 ca_certs: str | None = None, 

157 ca_cert_dir: str | None = None, 

158 ca_cert_data: None | str | bytes = None, 

159 ssl_minimum_version: int | None = None, 

160 ssl_maximum_version: int | None = None, 

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

162 cert_file: str | None = None, 

163 key_file: str | None = None, 

164 key_password: str | None = None, 

165 ) -> None: ...