Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/api_core/universe.py: 68%

1# Copyright 2024 Google LLC 

2# 

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

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

5# You may obtain a copy of the License at 

6# 

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

8# 

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

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

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

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

13# limitations under the License. 

14 

15"""Helpers for universe domain.""" 

16 

17from typing import Any, Optional 

18 

19DEFAULT_UNIVERSE = "googleapis.com" 

20 

21 

22class EmptyUniverseError(ValueError): 

23 def __init__(self): 

24 message = "Universe Domain cannot be an empty string." 

25 super().__init__(message) 

26 

27 

28class UniverseMismatchError(ValueError): 

29 def __init__(self, client_universe, credentials_universe): 

30 message = ( 

31 f"The configured universe domain ({client_universe}) does not match the universe domain " 

32 f"found in the credentials ({credentials_universe}). " 

33 "If you haven't configured the universe domain explicitly, " 

34 f"`{DEFAULT_UNIVERSE}` is the default." 

35 ) 

36 super().__init__(message) 

37 

38 

39def determine_domain( 

40 client_universe_domain: Optional[str], universe_domain_env: Optional[str] 

41) -> str: 

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

43 

44 Args: 

45 client_universe_domain (Optional[str]): The universe domain configured via the client options. 

46 universe_domain_env (Optional[str]): The universe domain configured via the 

47 "GOOGLE_CLOUD_UNIVERSE_DOMAIN" environment variable. 

48 

49 Returns: 

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

51 

52 Raises: 

53 ValueError: If the universe domain is an empty string. 

54 """ 

55 universe_domain = DEFAULT_UNIVERSE 

56 if client_universe_domain is not None: 

57 universe_domain = client_universe_domain 

58 elif universe_domain_env is not None: 

59 universe_domain = universe_domain_env 

60 if len(universe_domain.strip()) == 0: 

61 raise EmptyUniverseError 

62 return universe_domain 

63 

64 

65def compare_domains(client_universe: str, credentials: Any) -> bool: 

66 """Returns True iff the universe domains used by the client and credentials match. 

67 

68 Args: 

69 client_universe (str): The universe domain configured via the client options. 

70 credentials Any: The credentials being used in the client. 

71 

72 Returns: 

73 bool: True iff client_universe matches the universe in credentials. 

74 

75 Raises: 

76 ValueError: when client_universe does not match the universe in credentials. 

77 """ 

78 credentials_universe = getattr(credentials, "universe_domain", DEFAULT_UNIVERSE) 

79 

80 if client_universe != credentials_universe: 

81 raise UniverseMismatchError(client_universe, credentials_universe) 

82 return True