Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/iam_credentials_v1/services/iam_credentials/transports/rest_base.py: 52%

1# -*- coding: utf-8 -*- 

2# Copyright 2025 Google LLC 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15# 

16import json # type: ignore 

17import re 

18from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union 

19 

20from google.api_core import gapic_v1, path_template 

21from google.protobuf import json_format 

22 

23from google.cloud.iam_credentials_v1.types import common 

24 

25from .base import DEFAULT_CLIENT_INFO, IAMCredentialsTransport 

26 

27 

28class _BaseIAMCredentialsRestTransport(IAMCredentialsTransport): 

29 """Base REST backend transport for IAMCredentials. 

30 

31 Note: This class is not meant to be used directly. Use its sync and 

32 async sub-classes instead. 

33 

34 This class defines the same methods as the primary client, so the 

35 primary client can load the underlying transport implementation 

36 and call it. 

37 

38 It sends JSON representations of protocol buffers over HTTP/1.1 

39 """ 

40 

41 def __init__( 

42 self, 

43 *, 

44 host: str = "iamcredentials.googleapis.com", 

45 credentials: Optional[Any] = None, 

46 client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, 

47 always_use_jwt_access: Optional[bool] = False, 

48 url_scheme: str = "https", 

49 api_audience: Optional[str] = None, 

50 ) -> None: 

51 """Instantiate the transport. 

52 Args: 

53 host (Optional[str]): 

54 The hostname to connect to (default: 'iamcredentials.googleapis.com'). 

55 credentials (Optional[Any]): The 

56 authorization credentials to attach to requests. These 

57 credentials identify the application to the service; if none 

58 are specified, the client will attempt to ascertain the 

59 credentials from the environment. 

60 client_info (google.api_core.gapic_v1.client_info.ClientInfo): 

61 The client info used to send a user-agent string along with 

62 API requests. If ``None``, then default info will be used. 

63 Generally, you only need to set this if you are developing 

64 your own client library. 

65 always_use_jwt_access (Optional[bool]): Whether self signed JWT should 

66 be used for service account credentials. 

67 url_scheme: the protocol scheme for the API endpoint. Normally 

68 "https", but for testing or local servers, 

69 "http" can be specified. 

70 """ 

71 # Run the base constructor 

72 maybe_url_match = re.match("^(?P<scheme>http(?:s)?://)?(?P<host>.*)$", host) 

73 if maybe_url_match is None: 

74 raise ValueError( 

75 f"Unexpected hostname structure: {host}" 

76 ) # pragma: NO COVER 

77 

78 url_match_items = maybe_url_match.groupdict() 

79 

80 host = f"{url_scheme}://{host}" if not url_match_items["scheme"] else host 

81 

82 super().__init__( 

83 host=host, 

84 credentials=credentials, 

85 client_info=client_info, 

86 always_use_jwt_access=always_use_jwt_access, 

87 api_audience=api_audience, 

88 ) 

89 

90 class _BaseGenerateAccessToken: 

91 def __hash__(self): # pragma: NO COVER 

92 return NotImplementedError("__hash__ must be implemented.") 

93 

94 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

95 

96 @classmethod 

97 def _get_unset_required_fields(cls, message_dict): 

98 return { 

99 k: v 

100 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

101 if k not in message_dict 

102 } 

103 

104 @staticmethod 

105 def _get_http_options(): 

106 http_options: List[Dict[str, str]] = [ 

107 { 

108 "method": "post", 

109 "uri": "/v1/{name=projects/*/serviceAccounts/*}:generateAccessToken", 

110 "body": "*", 

111 }, 

112 ] 

113 return http_options 

114 

115 @staticmethod 

116 def _get_transcoded_request(http_options, request): 

117 pb_request = common.GenerateAccessTokenRequest.pb(request) 

118 transcoded_request = path_template.transcode(http_options, pb_request) 

119 return transcoded_request 

120 

121 @staticmethod 

122 def _get_request_body_json(transcoded_request): 

123 # Jsonify the request body 

124 

125 body = json_format.MessageToJson( 

126 transcoded_request["body"], use_integers_for_enums=True 

127 ) 

128 return body 

129 

130 @staticmethod 

131 def _get_query_params_json(transcoded_request): 

132 query_params = json.loads( 

133 json_format.MessageToJson( 

134 transcoded_request["query_params"], 

135 use_integers_for_enums=True, 

136 ) 

137 ) 

138 query_params.update( 

139 _BaseIAMCredentialsRestTransport._BaseGenerateAccessToken._get_unset_required_fields( 

140 query_params 

141 ) 

142 ) 

143 

144 query_params["$alt"] = "json;enum-encoding=int" 

145 return query_params 

146 

147 class _BaseGenerateIdToken: 

148 def __hash__(self): # pragma: NO COVER 

149 return NotImplementedError("__hash__ must be implemented.") 

150 

151 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

152 

153 @classmethod 

154 def _get_unset_required_fields(cls, message_dict): 

155 return { 

156 k: v 

157 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

158 if k not in message_dict 

159 } 

160 

161 @staticmethod 

162 def _get_http_options(): 

163 http_options: List[Dict[str, str]] = [ 

164 { 

165 "method": "post", 

166 "uri": "/v1/{name=projects/*/serviceAccounts/*}:generateIdToken", 

167 "body": "*", 

168 }, 

169 ] 

170 return http_options 

171 

172 @staticmethod 

173 def _get_transcoded_request(http_options, request): 

174 pb_request = common.GenerateIdTokenRequest.pb(request) 

175 transcoded_request = path_template.transcode(http_options, pb_request) 

176 return transcoded_request 

177 

178 @staticmethod 

179 def _get_request_body_json(transcoded_request): 

180 # Jsonify the request body 

181 

182 body = json_format.MessageToJson( 

183 transcoded_request["body"], use_integers_for_enums=True 

184 ) 

185 return body 

186 

187 @staticmethod 

188 def _get_query_params_json(transcoded_request): 

189 query_params = json.loads( 

190 json_format.MessageToJson( 

191 transcoded_request["query_params"], 

192 use_integers_for_enums=True, 

193 ) 

194 ) 

195 query_params.update( 

196 _BaseIAMCredentialsRestTransport._BaseGenerateIdToken._get_unset_required_fields( 

197 query_params 

198 ) 

199 ) 

200 

201 query_params["$alt"] = "json;enum-encoding=int" 

202 return query_params 

203 

204 class _BaseSignBlob: 

205 def __hash__(self): # pragma: NO COVER 

206 return NotImplementedError("__hash__ must be implemented.") 

207 

208 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

209 

210 @classmethod 

211 def _get_unset_required_fields(cls, message_dict): 

212 return { 

213 k: v 

214 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

215 if k not in message_dict 

216 } 

217 

218 @staticmethod 

219 def _get_http_options(): 

220 http_options: List[Dict[str, str]] = [ 

221 { 

222 "method": "post", 

223 "uri": "/v1/{name=projects/*/serviceAccounts/*}:signBlob", 

224 "body": "*", 

225 }, 

226 ] 

227 return http_options 

228 

229 @staticmethod 

230 def _get_transcoded_request(http_options, request): 

231 pb_request = common.SignBlobRequest.pb(request) 

232 transcoded_request = path_template.transcode(http_options, pb_request) 

233 return transcoded_request 

234 

235 @staticmethod 

236 def _get_request_body_json(transcoded_request): 

237 # Jsonify the request body 

238 

239 body = json_format.MessageToJson( 

240 transcoded_request["body"], use_integers_for_enums=True 

241 ) 

242 return body 

243 

244 @staticmethod 

245 def _get_query_params_json(transcoded_request): 

246 query_params = json.loads( 

247 json_format.MessageToJson( 

248 transcoded_request["query_params"], 

249 use_integers_for_enums=True, 

250 ) 

251 ) 

252 query_params.update( 

253 _BaseIAMCredentialsRestTransport._BaseSignBlob._get_unset_required_fields( 

254 query_params 

255 ) 

256 ) 

257 

258 query_params["$alt"] = "json;enum-encoding=int" 

259 return query_params 

260 

261 class _BaseSignJwt: 

262 def __hash__(self): # pragma: NO COVER 

263 return NotImplementedError("__hash__ must be implemented.") 

264 

265 __REQUIRED_FIELDS_DEFAULT_VALUES: Dict[str, Any] = {} 

266 

267 @classmethod 

268 def _get_unset_required_fields(cls, message_dict): 

269 return { 

270 k: v 

271 for k, v in cls.__REQUIRED_FIELDS_DEFAULT_VALUES.items() 

272 if k not in message_dict 

273 } 

274 

275 @staticmethod 

276 def _get_http_options(): 

277 http_options: List[Dict[str, str]] = [ 

278 { 

279 "method": "post", 

280 "uri": "/v1/{name=projects/*/serviceAccounts/*}:signJwt", 

281 "body": "*", 

282 }, 

283 ] 

284 return http_options 

285 

286 @staticmethod 

287 def _get_transcoded_request(http_options, request): 

288 pb_request = common.SignJwtRequest.pb(request) 

289 transcoded_request = path_template.transcode(http_options, pb_request) 

290 return transcoded_request 

291 

292 @staticmethod 

293 def _get_request_body_json(transcoded_request): 

294 # Jsonify the request body 

295 

296 body = json_format.MessageToJson( 

297 transcoded_request["body"], use_integers_for_enums=True 

298 ) 

299 return body 

300 

301 @staticmethod 

302 def _get_query_params_json(transcoded_request): 

303 query_params = json.loads( 

304 json_format.MessageToJson( 

305 transcoded_request["query_params"], 

306 use_integers_for_enums=True, 

307 ) 

308 ) 

309 query_params.update( 

310 _BaseIAMCredentialsRestTransport._BaseSignJwt._get_unset_required_fields( 

311 query_params 

312 ) 

313 ) 

314 

315 query_params["$alt"] = "json;enum-encoding=int" 

316 return query_params 

317 

318 

319__all__ = ("_BaseIAMCredentialsRestTransport",)