Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_vendor/cachecontrol/serialize.py: 23%

1# SPDX-FileCopyrightText: 2015 Eric Larson 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4from __future__ import annotations 

5 

6import io 

7from typing import IO, TYPE_CHECKING, Any, Mapping, cast 

8 

9from pip._vendor import msgpack 

10from pip._vendor.requests.structures import CaseInsensitiveDict 

11from pip._vendor.urllib3 import HTTPResponse 

12 

13if TYPE_CHECKING: 

14 from pip._vendor.requests import PreparedRequest 

15 

16 

17class Serializer: 

18 serde_version = "4" 

19 

20 def dumps( 

21 self, 

22 request: PreparedRequest, 

23 response: HTTPResponse, 

24 body: bytes | None = None, 

25 ) -> bytes: 

26 response_headers: CaseInsensitiveDict[str] = CaseInsensitiveDict( 

27 response.headers 

28 ) 

29 

30 if body is None: 

31 # When a body isn't passed in, we'll read the response. We 

32 # also update the response with a new file handler to be 

33 # sure it acts as though it was never read. 

34 body = response.read(decode_content=False) 

35 response._fp = io.BytesIO(body) # type: ignore[attr-defined] 

36 response.length_remaining = len(body) 

37 

38 data = { 

39 "response": { 

40 "body": body, # Empty bytestring if body is stored separately 

41 "headers": {str(k): str(v) for k, v in response.headers.items()}, # type: ignore[no-untyped-call] 

42 "status": response.status, 

43 "version": response.version, 

44 "reason": str(response.reason), 

45 "decode_content": response.decode_content, 

46 } 

47 } 

48 

49 # Construct our vary headers 

50 data["vary"] = {} 

51 if "vary" in response_headers: 

52 varied_headers = response_headers["vary"].split(",") 

53 for header in varied_headers: 

54 header = str(header).strip() 

55 header_value = request.headers.get(header, None) 

56 if header_value is not None: 

57 header_value = str(header_value) 

58 data["vary"][header] = header_value 

59 

60 return b",".join([f"cc={self.serde_version}".encode(), self.serialize(data)]) 

61 

62 def serialize(self, data: dict[str, Any]) -> bytes: 

63 return cast(bytes, msgpack.dumps(data, use_bin_type=True)) 

64 

65 def loads( 

66 self, 

67 request: PreparedRequest, 

68 data: bytes, 

69 body_file: IO[bytes] | None = None, 

70 ) -> HTTPResponse | None: 

71 # Short circuit if we've been given an empty set of data 

72 if not data: 

73 return None 

74 

75 # Determine what version of the serializer the data was serialized 

76 # with 

77 try: 

78 ver, data = data.split(b",", 1) 

79 except ValueError: 

80 ver = b"cc=0" 

81 

82 # Make sure that our "ver" is actually a version and isn't a false 

83 # positive from a , being in the data stream. 

84 if ver[:3] != b"cc=": 

85 data = ver + data 

86 ver = b"cc=0" 

87 

88 # Get the version number out of the cc=N 

89 verstr = ver.split(b"=", 1)[-1].decode("ascii") 

90 

91 # Dispatch to the actual load method for the given version 

92 try: 

93 return getattr(self, f"_loads_v{verstr}")(request, data, body_file) # type: ignore[no-any-return] 

94 

95 except AttributeError: 

96 # This is a version we don't have a loads function for, so we'll 

97 # just treat it as a miss and return None 

98 return None 

99 

100 def prepare_response( 

101 self, 

102 request: PreparedRequest, 

103 cached: Mapping[str, Any], 

104 body_file: IO[bytes] | None = None, 

105 ) -> HTTPResponse | None: 

106 """Verify our vary headers match and construct a real urllib3 

107 HTTPResponse object. 

108 """ 

109 # Special case the '*' Vary value as it means we cannot actually 

110 # determine if the cached response is suitable for this request. 

111 # This case is also handled in the controller code when creating 

112 # a cache entry, but is left here for backwards compatibility. 

113 if "*" in cached.get("vary", {}): 

114 return None 

115 

116 # Ensure that the Vary headers for the cached response match our 

117 # request 

118 for header, value in cached.get("vary", {}).items(): 

119 if request.headers.get(header, None) != value: 

120 return None 

121 

122 body_raw = cached["response"].pop("body") 

123 

124 headers: CaseInsensitiveDict[str] = CaseInsensitiveDict( 

125 data=cached["response"]["headers"] 

126 ) 

127 if headers.get("transfer-encoding", "") == "chunked": 

128 headers.pop("transfer-encoding") 

129 

130 cached["response"]["headers"] = headers 

131 

132 try: 

133 body: IO[bytes] 

134 if body_file is None: 

135 body = io.BytesIO(body_raw) 

136 else: 

137 body = body_file 

138 except TypeError: 

139 # This can happen if cachecontrol serialized to v1 format (pickle) 

140 # using Python 2. A Python 2 str(byte string) will be unpickled as 

141 # a Python 3 str (unicode string), which will cause the above to 

142 # fail with: 

143 # 

144 # TypeError: 'str' does not support the buffer interface 

145 body = io.BytesIO(body_raw.encode("utf8")) 

146 

147 # Discard any `strict` parameter serialized by older version of cachecontrol. 

148 cached["response"].pop("strict", None) 

149 

150 return HTTPResponse(body=body, preload_content=False, **cached["response"]) 

151 

152 def _loads_v0( 

153 self, 

154 request: PreparedRequest, 

155 data: bytes, 

156 body_file: IO[bytes] | None = None, 

157 ) -> None: 

158 # The original legacy cache data. This doesn't contain enough 

159 # information to construct everything we need, so we'll treat this as 

160 # a miss. 

161 return None 

162 

163 def _loads_v1( 

164 self, 

165 request: PreparedRequest, 

166 data: bytes, 

167 body_file: IO[bytes] | None = None, 

168 ) -> HTTPResponse | None: 

169 # The "v1" pickled cache format. This is no longer supported 

170 # for security reasons, so we treat it as a miss. 

171 return None 

172 

173 def _loads_v2( 

174 self, 

175 request: PreparedRequest, 

176 data: bytes, 

177 body_file: IO[bytes] | None = None, 

178 ) -> HTTPResponse | None: 

179 # The "v2" compressed base64 cache format. 

180 # This has been removed due to age and poor size/performance 

181 # characteristics, so we treat it as a miss. 

182 return None 

183 

184 def _loads_v3( 

185 self, 

186 request: PreparedRequest, 

187 data: bytes, 

188 body_file: IO[bytes] | None = None, 

189 ) -> None: 

190 # Due to Python 2 encoding issues, it's impossible to know for sure 

191 # exactly how to load v3 entries, thus we'll treat these as a miss so 

192 # that they get rewritten out as v4 entries. 

193 return None 

194 

195 def _loads_v4( 

196 self, 

197 request: PreparedRequest, 

198 data: bytes, 

199 body_file: IO[bytes] | None = None, 

200 ) -> HTTPResponse | None: 

201 try: 

202 cached = msgpack.loads(data, raw=False) 

203 except ValueError: 

204 return None 

205 

206 return self.prepare_response(request, cached, body_file)