Coverage for app/venv/lib/python3.14/site-packages/weblate/utils/environment.py: 55%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7import ast 

8import os 

9from typing import Any 

10from urllib.parse import quote 

11 

12from django.core.exceptions import ImproperlyConfigured 

13 

14 

15def get_env_str( 

16 name: str, 

17 default: str | None = None, 

18 required: bool = False, 

19 fallback_name: str | None = None, 

20) -> str: 

21 file_env = f"{name}_FILE" 

22 if filename := os.environ.get(file_env): 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 try: 

24 with open(filename) as handle: 

25 result = handle.read() 

26 except OSError as error: 

27 msg = f"Failed to open {filename} as specified by {file_env}: {error}" 

28 raise ImproperlyConfigured(msg) from error 

29 else: 

30 if fallback_name and name not in os.environ: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

31 name = fallback_name 

32 result = os.environ.get( 

33 name, 

34 default, # type: ignore[arg-type] 

35 ) 

36 if required and not result: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 msg = f"{name} has to be configured!" 

38 raise ImproperlyConfigured(msg) 

39 return result 

40 

41 

42def get_env_list(name: str, default: list[str] | None = None) -> list[str]: 

43 """Get list from environment.""" 

44 if name not in os.environ: 

45 return default or [] 

46 return os.environ[name].split(",") 

47 

48 

49def get_env_map(name: str, default: dict[str, str] | None = None) -> dict[str, str]: 

50 """ 

51 Get mapping from environment. 

52 

53 parses 'full_name:name,email:mail' into {'email': 'mail', 'full_name': 'name'} 

54 """ 

55 if os.environ.get(name): 

56 return dict(e.split(":") for e in os.environ[name].split(",")) 

57 return default or {} 

58 

59 

60def get_env_int(name: str, default: int = 0) -> int: 

61 """Get integer value from environment.""" 

62 if name not in os.environ: 

63 return default 

64 try: 

65 return int(os.environ[name]) 

66 except ValueError as error: 

67 msg = f"{name} is not an integer: {error}" 

68 raise ImproperlyConfigured(msg) from error 

69 

70 

71def get_env_float(name: str, default: float = 0.0) -> float: 

72 """Get float value from environment.""" 

73 if name not in os.environ: 73 ↛ 75line 73 didn't jump to line 75 because the condition on line 73 was always true

74 return default 

75 try: 

76 return float(os.environ[name]) 

77 except ValueError as error: 

78 msg = f"{name} is not an float: {error}" 

79 raise ImproperlyConfigured(msg) from error 

80 

81 

82def get_env_bool(name: str, default: bool = False) -> bool: 

83 """Get boolean value from environment.""" 

84 if name not in os.environ: 

85 return default 

86 true_values = {"true", "yes", "1"} 

87 return os.environ[name].lower() in true_values 

88 

89 

90def modify_env_list(current: list[str], name: str) -> list[str]: 

91 """Modify list based on environment (for example checks).""" 

92 for item in reversed(get_env_list(f"WEBLATE_ADD_{name}")): 92 ↛ 93line 92 didn't jump to line 93 because the loop on line 92 never started

93 current.insert(0, item) 

94 for item in get_env_list(f"WEBLATE_REMOVE_{name}"): 94 ↛ 95line 94 didn't jump to line 95 because the loop on line 94 never started

95 current.remove(item) 

96 return current 

97 

98 

99def get_env_credentials( 

100 name: str, 

101) -> dict[str, dict[str, str]]: 

102 """Get VCS integration credentials from environment.""" 

103 if found_env_credentials := get_env_str(f"WEBLATE_{name}_CREDENTIALS"): 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true

104 try: 

105 return ast.literal_eval(found_env_credentials) 

106 except ValueError as error: 

107 msg = f"Could not parse {name}_CREDENTIALS: {error}" 

108 raise ImproperlyConfigured(msg) from error 

109 

110 username = os.environ.get(f"WEBLATE_{name}_USERNAME", "") 

111 token = os.environ.get(f"WEBLATE_{name}_TOKEN", "") 

112 host = os.environ.get(f"WEBLATE_{name}_HOST") 

113 organization = os.environ.get(f"WEBLATE_{name}_ORGANIZATION") 

114 

115 if not host: 115 ↛ 121line 115 didn't jump to line 121 because the condition on line 115 was always true

116 if username or token: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true

117 msg = f"Incomplete {name}_CREDENTIALS configuration: missing WEBLATE_{name}_HOST" 

118 raise ImproperlyConfigured(msg) 

119 return {} 

120 

121 credentials = {host: {"username": username, "token": token}} 

122 

123 if organization is not None: 

124 credentials[host]["organization"] = organization 

125 

126 return credentials 

127 

128 

129def get_env_ratelimit(name: str, default: str) -> str: 

130 value = os.environ.get(name, default) 

131 

132 # Taken from rest_framework.throttling.SimpleRateThrottle.parse_rate 

133 # it can not be imported here as that breaks config loading for 

134 # rest_framework 

135 

136 try: 

137 num, period = value.split("/") 

138 except ValueError as error: 

139 msg = f"Could not parse {name}: {error}" 

140 raise ImproperlyConfigured(msg) from error 

141 if not num.isdigit(): 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true

142 msg = f"Could not parse {name}: rate is not numeric: {num}" 

143 raise ImproperlyConfigured(msg) 

144 if period[0] not in {"s", "m", "h", "d"}: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 msg = f"Could not parse {name}: unknown period: {period}" 

146 raise ImproperlyConfigured(msg) 

147 

148 return value 

149 

150 

151def url_quote_part(value: str) -> str: 

152 return quote(value, safe="") 

153 

154 

155def get_env_redis_url() -> str: 

156 # Get values from the environment 

157 redis_proto = "rediss" if get_env_bool("REDIS_TLS") else "redis" 

158 redis_host = url_quote_part(get_env_str("REDIS_HOST", "cache", required=True)) 

159 redis_port = get_env_int("REDIS_PORT", 6379) 

160 redis_db = get_env_int("REDIS_DB", 1) 

161 redis_password = url_quote_part(get_env_str("REDIS_PASSWORD", "")) 

162 redis_user = url_quote_part(get_env_str("REDIS_USER", "")) 

163 

164 # Build user/password part of the URL 

165 redis_user_password: str | None 

166 if redis_user and redis_password: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true

167 redis_user_password = f"{redis_user}:{redis_password}@" 

168 elif redis_password: 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true

169 redis_user_password = f":{redis_password}@" 

170 elif redis_user: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true

171 redis_user_password = f"{redis_user}@" 

172 else: 

173 redis_user_password = "" 

174 

175 return f"{redis_proto}://{redis_user_password}{redis_host}:{redis_port}/{redis_db}" 

176 

177 

178def get_saml_idp() -> dict[str, Any] | None: 

179 idp_entity_id = get_env_str("WEBLATE_SAML_IDP_ENTITY_ID") 

180 if not idp_entity_id: 180 ↛ 183line 180 didn't jump to line 183 because the condition on line 180 was always true

181 return None 

182 # Identity Provider 

183 saml_idp = { 

184 "entity_id": idp_entity_id, 

185 "url": get_env_str("WEBLATE_SAML_IDP_URL"), 

186 "x509cert": get_env_str("WEBLATE_SAML_IDP_X509CERT"), 

187 } 

188 

189 for field in ( 

190 "attr_full_name", 

191 "attr_first_name", 

192 "attr_last_name", 

193 "attr_username", 

194 "attr_email", 

195 "attr_user_permanent_id", 

196 ): 

197 env_name = f"WEBLATE_SAML_ID_{field.upper()}" 

198 value = get_env_str(env_name) 

199 if value: 

200 saml_idp[field] = value 

201 

202 return saml_idp