Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/secretmanager_v1beta1/_compat.py: 93%

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

2# Copyright 2026 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# 

16"""A compatibility module for older versions of google-api-core.""" 

17 

18import json 

19import os 

20from typing import Any, Dict, List, Optional, Tuple 

21from urllib.parse import urlparse, urlunparse 

22 

23from google.api_core import path_template 

24from google.api_core.universe import EmptyUniverseError 

25from google.auth.exceptions import MutualTLSChannelError 

26from google.protobuf import json_format 

27 

28try: 

29 # note: `#type: ignore` is added because the return type for `should_use_client_cert` 

30 # is different than that of the fallback implementation below. This will be removed once 

31 # we bump the minimum supported version of google-auth. 

32 from google.auth.transport.mtls import should_use_client_cert # type: ignore 

33except ImportError: # pragma: NO COVER 

34 

35 def should_use_client_cert(): 

36 """Returns whether client certificate should be used for mTLS.""" 

37 use_client_cert = os.getenv( 

38 "GOOGLE_API_USE_CLIENT_CERTIFICATE", "false" 

39 ).lower() 

40 if use_client_cert not in ("true", "false"): 

41 raise ValueError( 

42 "Environment variable `GOOGLE_API_USE_CLIENT_CERTIFICATE` must be" 

43 " either `true` or `false`" 

44 ) 

45 return use_client_cert == "true" 

46 

47 

48def read_environment_variables(): 

49 """Returns the environment variables used by the client. 

50 

51 Returns: 

52 Tuple[bool, str, str]: returns the GOOGLE_API_USE_CLIENT_CERTIFICATE, 

53 GOOGLE_API_USE_MTLS_ENDPOINT, and GOOGLE_CLOUD_UNIVERSE_DOMAIN environment variables. 

54 

55 Raises: 

56 ValueError: If GOOGLE_API_USE_CLIENT_CERTIFICATE is not 

57 any of ["true", "false"]. 

58 google.auth.exceptions.MutualTLSChannelError: If GOOGLE_API_USE_MTLS_ENDPOINT 

59 is not any of ["auto", "never", "always"]. 

60 """ 

61 use_client_cert = should_use_client_cert() 

62 use_mtls_endpoint = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto").lower() 

63 universe_domain_env = os.getenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN") 

64 if use_mtls_endpoint not in ("auto", "never", "always"): 

65 raise MutualTLSChannelError( 

66 "Environment variable `GOOGLE_API_USE_MTLS_ENDPOINT` must be `never`," 

67 " `auto` or `always`" 

68 ) 

69 return use_client_cert, use_mtls_endpoint, universe_domain_env 

70 

71 

72DEFAULT_UNIVERSE = "googleapis.com" 

73 

74try: 

75 from google.api_core.universe import get_default_mtls_endpoint 

76except ImportError: # pragma: NO COVER 

77 

78 def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: 

79 """Converts api endpoint to mTLS endpoint. 

80 

81 Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to 

82 "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. 

83 Other URLs (including those that do not match these domain suffixes or 

84 already contain '.mtls.') are passed through as-is. 

85 

86 Args: 

87 api_endpoint (Optional[str]): the api endpoint to convert. 

88 

89 Returns: 

90 Optional[str]: converted mTLS api endpoint. 

91 """ 

92 if not api_endpoint or ".mtls." in api_endpoint.lower(): 

93 return api_endpoint 

94 

95 has_scheme = "://" in api_endpoint 

96 if not has_scheme: 

97 parsed = urlparse("//" + api_endpoint) 

98 else: 

99 parsed = urlparse(api_endpoint) 

100 

101 host = parsed.hostname 

102 if not host: 

103 return api_endpoint 

104 

105 port = f":{parsed.port}" if parsed.port else "" 

106 

107 lowered_host = host.lower() 

108 suffix_sandbox = ".sandbox.googleapis.com" 

109 suffix_google = ".googleapis.com" 

110 if lowered_host.endswith(suffix_sandbox): 

111 new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" 

112 elif lowered_host.endswith(suffix_google): 

113 new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" 

114 else: 

115 return api_endpoint 

116 

117 netloc = new_host + port 

118 new_parsed = parsed._replace(netloc=netloc) 

119 

120 if not has_scheme: 

121 return urlunparse(new_parsed)[2:] 

122 else: 

123 return urlunparse(new_parsed) 

124 

125 

126try: 

127 from google.api_core.universe import get_api_endpoint 

128except ImportError: # pragma: NO COVER 

129 

130 def get_api_endpoint( 

131 api_override: Optional[str], 

132 universe_domain: str, 

133 default_universe: str, 

134 default_mtls_endpoint: Optional[str], 

135 default_endpoint_template: str, 

136 use_mtls: bool, 

137 ) -> str: 

138 """Return the API endpoint used by the client. 

139 

140 Args: 

141 api_override (Optional[str]): The API endpoint override. If specified, 

142 this is always returned. 

143 universe_domain (str): The universe domain used by the client. 

144 default_universe (str): The default universe domain. 

145 default_mtls_endpoint (Optional[str]): The default mTLS endpoint. 

146 default_endpoint_template (str): The default endpoint template containing 

147 a placeholder `{UNIVERSE_DOMAIN}`. 

148 use_mtls (bool): Whether to use the mTLS endpoint. 

149 

150 Returns: 

151 str: The API endpoint to be used by the client. 

152 

153 Raises: 

154 google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but 

155 not supported in the configured universe domain. 

156 ValueError: If mTLS is requested but no mTLS endpoint is available. 

157 """ 

158 if api_override is not None: 

159 return api_override 

160 

161 if use_mtls: 

162 if universe_domain.lower() != default_universe.lower(): 

163 raise MutualTLSChannelError( 

164 f"mTLS is not supported in any universe other than {default_universe}." 

165 ) 

166 if not default_mtls_endpoint: 

167 raise ValueError("mTLS endpoint is not available.") 

168 return default_mtls_endpoint 

169 else: 

170 return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain) 

171 

172 

173try: 

174 from google.api_core.universe import get_universe_domain 

175except ImportError: # pragma: NO COVER 

176 

177 def get_universe_domain( 

178 *potential_universes: Optional[str], 

179 default_universe: str, 

180 ) -> str: 

181 """Return the universe domain used by the client. 

182 

183 Args: 

184 *potential_universes (Optional[str]): Potential universe domains in order of preference. 

185 default_universe (str): The default universe domain. 

186 

187 Returns: 

188 str: The universe domain to be used by the client. 

189 

190 Raises: 

191 EmptyUniverseError: If the resolved universe domain is an empty string. 

192 """ 

193 resolved = next( 

194 (x.strip() for x in potential_universes if x is not None), 

195 default_universe, 

196 ) 

197 

198 if not resolved: 

199 raise EmptyUniverseError() 

200 return resolved 

201 

202 

203try: 

204 from google.api_core.rest_helpers import transcode_request # type: ignore 

205except ImportError: # pragma: NO COVER 

206 

207 def transcode_request( 

208 http_options: List[Dict[str, str]], 

209 request: Any, 

210 required_fields_default_values: Optional[Dict[str, Any]] = None, 

211 rest_numeric_enums: bool = False, 

212 ) -> Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: 

213 """Transcodes a request into HTTP method, URI, body, and query parameters. 

214 

215 Args: 

216 http_options (List[Dict[str, str]]): List of HTTP transcoding rules. 

217 request (Any): The protobuf or proto-plus request message. 

218 required_fields_default_values (Optional[Dict[str, Any]]): Dictionary 

219 of required fields default values to merge into query parameters if missing. 

220 rest_numeric_enums (bool): Whether to encode enums as integers. 

221 

222 Returns: 

223 Tuple[Dict[str, Any], Optional[str], Dict[str, Any]]: A tuple containing: 

224 - The raw transcoded request dictionary (containing keys like 'uri', 'method'). 

225 - The serialized request body JSON string, or None if no body. 

226 - The query parameters dictionary. 

227 """ 

228 if request is None: 

229 raise TypeError("request cannot be None") 

230 

231 # Convert proto-plus message to its underlying protobuf message if needed 

232 pb_request = getattr(request, "_pb", request) 

233 

234 transcoded_request = path_template.transcode(http_options, pb_request) 

235 

236 body_json = None 

237 if transcoded_request.get("body") is not None: 

238 body_json = json_format.MessageToJson( 

239 transcoded_request["body"], 

240 use_integers_for_enums=rest_numeric_enums, 

241 ) 

242 

243 query_params_json = {} 

244 if transcoded_request.get("query_params") is not None: 

245 query_params_json = json.loads( 

246 json_format.MessageToJson( 

247 transcoded_request["query_params"], 

248 use_integers_for_enums=rest_numeric_enums, 

249 ) 

250 ) 

251 

252 # If required_fields_default_values is provided, we merge default values for missing 

253 # required fields into the query parameters. 

254 if required_fields_default_values: 

255 for k, v in required_fields_default_values.items(): 

256 if k not in query_params_json: 

257 query_params_json[k] = v 

258 

259 if rest_numeric_enums: 

260 query_params_json["$alt"] = "json;enum-encoding=int" 

261 

262 return transcoded_request, body_json, query_params_json