Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/core_api/services/ui/connections.py: 25%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

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

10# 

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

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17 

18from __future__ import annotations 

19 

20import logging 

21from collections.abc import MutableMapping 

22from functools import cache 

23from typing import TYPE_CHECKING, Literal 

24 

25from airflow.api_fastapi.core_api.datamodels.connections import ( 

26 ConnectionHookFieldBehavior, 

27 ConnectionHookMetaData, 

28 StandardHookFields, 

29) 

30from airflow.providers_manager import HookInfo, ProvidersManager 

31from airflow.serialization.definitions.param import SerializedParam 

32 

33if TYPE_CHECKING: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 from airflow.providers_manager import ConnectionFormWidgetInfo 

35 

36log = logging.getLogger(__name__) 

37 

38 

39class HookMetaService: 

40 """Service for retrieving details about hooks to render UI.""" 

41 

42 class MockOptional: 

43 """Mock for wtforms.validators.Optional.""" 

44 

45 def __init__( 

46 self, 

47 *args, 

48 **kwargs, 

49 ): 

50 pass 

51 

52 def __call__(self, form, field): 

53 """No-op call to satisfy WTForms validator protocol.""" 

54 return None 

55 

56 class MockEnum: 

57 """Mock for wtforms.validators.AnyOf.""" 

58 

59 def __init__(self, allowed_values): 

60 self.allowed_values = allowed_values 

61 

62 class MockBaseField: 

63 """Mock of WTForms Field.""" 

64 

65 param_type: str = "UNDEFINED" 

66 param_format: str | None = None 

67 widget = None 

68 

69 def __init__( 

70 self, 

71 label: str | None = None, 

72 validators=None, 

73 description: str = "", 

74 default: str | None = None, 

75 widget=None, 

76 source: Literal["dag", "task"] | None = None, 

77 ): 

78 type: str | list[str] = [self.param_type, "null"] 

79 enum = {} 

80 format = {"format": self.param_format} if self.param_format else {} 

81 if validators: 

82 if any(isinstance(v, HookMetaService.MockOptional) for v in validators): 

83 type = [self.param_type, "null"] 

84 for v in validators: 

85 if isinstance(v, HookMetaService.MockEnum): 

86 enum = {"enum": v.allowed_values} 

87 self.param = SerializedParam( 

88 default=default, 

89 title=label, 

90 description=description or None, 

91 source=source or None, 

92 type=type, 

93 **format, 

94 **enum, 

95 ) 

96 self.widget = widget 

97 self.field_class = self.__class__ 

98 

99 class MockStringField(MockBaseField): 

100 """Mock of WTForms StringField.""" 

101 

102 param_type: str = "string" 

103 

104 class MockIntegerField(MockBaseField): 

105 """Mock of WTForms IntegerField.""" 

106 

107 param_type: str = "integer" 

108 

109 class MockPasswordField(MockBaseField): 

110 """Mock of WTForms PasswordField.""" 

111 

112 param_type: str = "string" 

113 param_format: str | None = "password" 

114 

115 class MockBooleanField(MockBaseField): 

116 """Mock of WTForms BooleanField.""" 

117 

118 param_type: str = "boolean" 

119 

120 class MockAnyWidget: 

121 """Mock any flask appbuilder widget.""" 

122 

123 @staticmethod 

124 def _get_hooks_with_mocked_fab() -> tuple[ 

125 MutableMapping[str, HookInfo | None], dict[str, ConnectionFormWidgetInfo], dict[str, dict] 

126 ]: 

127 """Get hooks with all details w/o FAB needing to be installed.""" 

128 from unittest import mock 

129 

130 def mock_lazy_gettext(txt: str) -> str: 

131 """Mock for flask_babel.lazy_gettext.""" 

132 return txt 

133 

134 def mock_any_of(allowed_values: list) -> HookMetaService.MockEnum: 

135 """Mock for wtforms.validators.any_of.""" 

136 return HookMetaService.MockEnum(allowed_values) 

137 

138 # Before importing ProvidersManager, we need to mock all FAB and WTForms 

139 # dependencies to avoid ImportErrors when FAB is not installed. 

140 import sys 

141 from importlib.util import find_spec 

142 from unittest.mock import MagicMock 

143 

144 for mod_name in [ 

145 "wtforms", 

146 "wtforms.csrf", 

147 "wtforms.fields", 

148 "wtforms.fields.simple", 

149 "wtforms.validators", 

150 "flask_babel", 

151 "flask_appbuilder", 

152 "flask_appbuilder.fieldwidgets", 

153 ]: 

154 try: 

155 if not find_spec(mod_name): 

156 raise ModuleNotFoundError(f"No module named {mod_name!r}", name=mod_name) 

157 except ModuleNotFoundError: 

158 sys.modules[mod_name] = MagicMock() 

159 

160 # We conditionally inject mock classes for missing dependencies 

161 # to ensure `ProvidersManager` can initialize hook connection widgets 

162 # without crashing when FAB/WTForms are not installed. 

163 if isinstance(sys.modules.get("wtforms"), MagicMock): 

164 # Only apply mocks if the actual module wasn't loaded beforehand. 

165 # This avoids thread-safety issues caused by `unittest.mock.patch` mutating global states. 

166 with ( 

167 mock.patch("wtforms.StringField", HookMetaService.MockStringField), 

168 mock.patch("wtforms.fields.StringField", HookMetaService.MockStringField), 

169 mock.patch("wtforms.fields.simple.StringField", HookMetaService.MockStringField), 

170 mock.patch("wtforms.IntegerField", HookMetaService.MockIntegerField), 

171 mock.patch("wtforms.fields.IntegerField", HookMetaService.MockIntegerField), 

172 mock.patch("wtforms.PasswordField", HookMetaService.MockPasswordField), 

173 mock.patch("wtforms.BooleanField", HookMetaService.MockBooleanField), 

174 mock.patch("wtforms.fields.BooleanField", HookMetaService.MockBooleanField), 

175 mock.patch("wtforms.fields.simple.BooleanField", HookMetaService.MockBooleanField), 

176 mock.patch("flask_babel.lazy_gettext", mock_lazy_gettext), 

177 mock.patch("flask_appbuilder.fieldwidgets.BS3TextFieldWidget", HookMetaService.MockAnyWidget), 

178 mock.patch( 

179 "flask_appbuilder.fieldwidgets.BS3TextAreaFieldWidget", HookMetaService.MockAnyWidget 

180 ), 

181 mock.patch( 

182 "flask_appbuilder.fieldwidgets.BS3PasswordFieldWidget", HookMetaService.MockAnyWidget 

183 ), 

184 mock.patch("wtforms.validators.Optional", HookMetaService.MockOptional), 

185 mock.patch("wtforms.validators.any_of", mock_any_of), 

186 # Prevent poisoning the global ProvidersManager singleton with mocks 

187 mock.patch("airflow.providers_manager.ProvidersManager._instance", None), 

188 mock.patch("airflow.providers_manager.ProvidersManager.initialized", return_value=False), 

189 ): 

190 pm = ProvidersManager() 

191 return pm.hooks, pm.connection_form_widgets, pm.field_behaviours # Will init providers hooks 

192 else: 

193 pm = ProvidersManager() 

194 return pm.hooks, pm.connection_form_widgets, pm.field_behaviours # Will init providers hooks 

195 

196 @staticmethod 

197 def _make_standard_fields(field_behaviour: dict | None) -> StandardHookFields | None: 

198 if not field_behaviour: 

199 return None 

200 

201 def make_field(field_name: str, field_behaviour: dict) -> ConnectionHookFieldBehavior | None: 

202 hidden_fields = field_behaviour.get("hidden_fields", []) 

203 relabeling = field_behaviour.get("relabeling", {}).get(field_name) 

204 placeholder = field_behaviour.get("placeholders", {}).get(field_name) 

205 if any([field_name in hidden_fields, relabeling, placeholder]): 

206 return ConnectionHookFieldBehavior( 

207 hidden=field_name in hidden_fields, 

208 title=relabeling, 

209 placeholder=placeholder, 

210 ) 

211 return None 

212 

213 return StandardHookFields( 

214 description=make_field("description", field_behaviour), 

215 url_schema=make_field("schema", field_behaviour), 

216 host=make_field("host", field_behaviour), 

217 port=make_field("port", field_behaviour), 

218 login=make_field("login", field_behaviour), 

219 password=make_field("password", field_behaviour), 

220 ) 

221 

222 @staticmethod 

223 def _convert_extra_fields(form_widgets: dict[str, ConnectionFormWidgetInfo]) -> dict[str, MutableMapping]: 

224 result: dict[str, MutableMapping] = {} 

225 for key, form_widget in form_widgets.items(): 

226 hook_key = key.split("__")[1] 

227 hook_widgets = result.get(hook_key, {}) 

228 

229 if isinstance(form_widget.field, dict): 

230 # yaml path, form widgets read from yaml and already present in SerializedParam.dump() format 

231 hook_widgets[form_widget.field_name] = form_widget.field 

232 elif isinstance(form_widget.field, HookMetaService.MockBaseField): 

233 # legacy path, form widgets created using mocked WTForms fields, need to convert to SerializedParam.dump() 

234 hook_widgets[form_widget.field_name] = form_widget.field.param.dump() 

235 elif type(form_widget.field).__name__ == "UnboundField": 

236 # handle real WTForms fields gracefully without needing mock patches 

237 field_class_name = getattr(form_widget.field.field_class, "__name__", "") 

238 param_type = "string" 

239 param_format = None 

240 if field_class_name == "BooleanField": 

241 param_type = "boolean" 

242 elif field_class_name == "IntegerField": 

243 param_type = "integer" 

244 elif field_class_name == "PasswordField": 

245 param_format = "password" 

246 

247 label = ( 

248 form_widget.field.args[0] 

249 if len(form_widget.field.args) > 0 

250 else form_widget.field.kwargs.get("label") 

251 ) 

252 validators = form_widget.field.kwargs.get("validators", []) 

253 description = form_widget.field.kwargs.get("description", "") 

254 default = form_widget.field.kwargs.get("default", None) 

255 

256 enum = {} 

257 for v in validators: 

258 if type(v).__name__ == "AnyOf": 

259 enum["enum"] = getattr(v, "values", []) 

260 

261 types = [param_type, "null"] 

262 format_dict = {"format": param_format} if param_format else {} 

263 

264 param = SerializedParam( 

265 default=default, 

266 title=str(label) if label is not None else None, 

267 description=str(description) if description else None, 

268 source=None, 

269 type=types, 

270 **format_dict, 

271 **enum, 

272 ).dump() 

273 hook_widgets[form_widget.field_name] = param 

274 else: 

275 log.error("Unknown form widget in %s: %s", hook_key, form_widget) 

276 continue 

277 

278 result[hook_key] = hook_widgets 

279 return result 

280 

281 @staticmethod 

282 @cache 

283 def hook_meta_data() -> list[ConnectionHookMetaData]: 

284 pm = ProvidersManager() 

285 widgets = HookMetaService._convert_extra_fields(pm._connection_form_widgets_from_metadata) 

286 return [ 

287 ConnectionHookMetaData( 

288 connection_type=meta.connection_type, 

289 hook_class_name=meta.hook_class_name, 

290 default_conn_name=None, 

291 hook_name=meta.hook_name, 

292 standard_fields=HookMetaService._make_standard_fields(meta.field_behaviour), 

293 extra_fields=widgets.get(meta.connection_type), 

294 ) 

295 for meta in pm.iter_connection_type_hook_ui_metadata() 

296 ]