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

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# 

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 @staticmethod 

97 def _get_http_options(): 

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

99 { 

100 "method": "post", 

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

102 "body": "*", 

103 }, 

104 ] 

105 return http_options 

106 

107 class _BaseGenerateIdToken: 

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

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

110 

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

112 

113 @staticmethod 

114 def _get_http_options(): 

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

116 { 

117 "method": "post", 

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

119 "body": "*", 

120 }, 

121 ] 

122 return http_options 

123 

124 class _BaseSignBlob: 

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

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

127 

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

129 

130 @staticmethod 

131 def _get_http_options(): 

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

133 { 

134 "method": "post", 

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

136 "body": "*", 

137 }, 

138 ] 

139 return http_options 

140 

141 class _BaseSignJwt: 

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

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

144 

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

146 

147 @staticmethod 

148 def _get_http_options(): 

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

150 { 

151 "method": "post", 

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

153 "body": "*", 

154 }, 

155 ] 

156 return http_options 

157 

158 

159__all__ = ("_BaseIAMCredentialsRestTransport",)