10
10
@pytest .mark .parametrize (
11
11
"rule, source, expected" ,
12
12
[
13
- (
13
+ pytest . param (
14
14
UseFunctionLevelImports ,
15
- # Test that all required imports within functions are being added to functions.
16
15
"""
17
16
from functools import reduce
18
17
from operator import add
@@ -42,10 +41,10 @@ def other_thing():
42
41
import dataclass
43
42
return dataclass(something(1, 2))
44
43
""" ,
44
+ id = "UseFunctionLevelImports | SUCCESS | general" ,
45
45
),
46
- (
46
+ pytest . param (
47
47
UseFunctionLevelImports ,
48
- # Test that it skips the dag decoratored functions.
49
48
"""
50
49
from airflow.decorators import dag, task
51
50
from operator import add
@@ -67,10 +66,10 @@ def my_task():
67
66
from operator import add
68
67
add(1, 2)
69
68
""" ,
69
+ id = "UseFunctionLevelImports | SKIP | dag decorator" ,
70
70
),
71
- (
71
+ pytest . param (
72
72
UseFunctionLevelImports ,
73
- # Test that it only adds unique imports i.e. only once
74
73
"""
75
74
from airflow.decorators import dag, task
76
75
from operator import add
@@ -94,10 +93,10 @@ def my_task():
94
93
add(1, 2)
95
94
add(1, 2)
96
95
""" ,
96
+ id = "UseFunctionLevelImports | SUCCESS | unique" ,
97
97
),
98
- (
98
+ pytest . param (
99
99
UseFunctionLevelImports ,
100
- # Test that it ignores imports for function decorators.
101
100
"""
102
101
import random
103
102
from airflow import DAG
@@ -119,10 +118,10 @@ def random_choice():
119
118
import random
120
119
return random.choice(['task_1', 'task_2'])
121
120
""" ,
121
+ id = "UseFunctionLevelImports | SKIP | function decorators" ,
122
122
),
123
- (
123
+ pytest . param (
124
124
UseJinjaVariableGet ,
125
- # Test that direct assignment of Variable.get is being transformed to jinja equivalent.
126
125
"""
127
126
from airflow.models import Variable
128
127
from operators.fake import FakeOperator
@@ -135,10 +134,10 @@ def random_choice():
135
134
136
135
FakeOperator(task_id="fake", foo='{{ var.value.FOO }}')
137
136
""" ,
137
+ id = "UseJinjaVariableGet | SUCCESS | direct assignment with key" ,
138
138
),
139
- (
139
+ pytest . param (
140
140
UseJinjaVariableGet ,
141
- # Test that nothing happens if it cannot import the module.
142
141
"""
143
142
from airflow.models import Variable
144
143
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
@@ -151,10 +150,10 @@ def random_choice():
151
150
152
151
KubernetesPodOperator(task_id="fake", image=Variable.get("FOO"))
153
152
""" ,
153
+ id = "UseJinjaVariableGet | SKIP | cannot import module" ,
154
154
),
155
- (
155
+ pytest . param (
156
156
UseJinjaVariableGet ,
157
- # Test that nothing happens if the import cannot be reached.
158
157
"""
159
158
from airflow.models import Variable
160
159
@@ -171,10 +170,10 @@ def foo():
171
170
172
171
FakeOperator(task_id="fake", foo=Variable.get("FOO"))
173
172
""" ,
173
+ id = "UseJinjaVariableGet | SKIP | cannot reach import" ,
174
174
),
175
- (
175
+ pytest . param (
176
176
UseJinjaVariableGet ,
177
- # Test that nothing happens if it is not in template_fields.
178
177
"""
179
178
from airflow.models import Variable
180
179
from operators.fake import FakeOperator
@@ -187,10 +186,10 @@ def foo():
187
186
188
187
FakeOperator(task_id="fake", fizz=Variable.get("FOO"))
189
188
""" ,
189
+ id = "UseJinjaVariableGet | SKIP | keyword not in template_fields" ,
190
190
),
191
- (
191
+ pytest . param (
192
192
UseJinjaVariableGet ,
193
- # Test that variable assignment of Variable.get is being transformed to jinja equivalent.
194
193
"""
195
194
from airflow.models import Variable
196
195
from operators.fake import FakeOperator
@@ -207,10 +206,10 @@ def foo():
207
206
208
207
FakeOperator(task_id="fake", foo=var)
209
208
""" ,
209
+ id = "UseJinjaVariableGet | SUCCESS | variable assignment" ,
210
210
),
211
- (
211
+ pytest . param (
212
212
UseJinjaVariableGet ,
213
- # Test that nothing happens if the variable cannot be reached.
214
213
"""
215
214
from airflow.models import Variable
216
215
from operators.fake import FakeOperator
@@ -229,10 +228,10 @@ def foo():
229
228
230
229
FakeOperator(task_id="fake", foo=var)
231
230
""" ,
231
+ id = "UseJinjaVariableGet | SKIP | cannot reach variable" ,
232
232
),
233
- (
233
+ pytest . param (
234
234
UseJinjaVariableGet ,
235
- # Test that variable assignment works for multiple keywords.
236
235
"""
237
236
from airflow.models import Variable
238
237
from operators.fake import FakeOperator
@@ -249,10 +248,10 @@ def foo():
249
248
250
249
FakeOperator(task_id="fake", foo=var, bar=var)
251
250
""" ,
251
+ id = "UseJinjaVariableGet | SUCCESS | variable assignment with multiple keywords" ,
252
252
),
253
- (
253
+ pytest . param (
254
254
UseJinjaVariableGet ,
255
- # Test that nothing happens if at least one keyword is not in template_fields.
256
255
"""
257
256
from airflow.models import Variable
258
257
from operators.fake import FakeOperator
@@ -269,10 +268,10 @@ def foo():
269
268
270
269
FakeOperator(task_id="fake", foo=var, fizz=var)
271
270
""" ,
271
+ id = "UseJinjaVariableGet | SKIP | variable assignment at least one keyword not in template_fields" ,
272
272
),
273
- (
273
+ pytest . param (
274
274
UseJinjaVariableGet ,
275
- # Test that nothing happens if variable is being referenced in multiple Calls where at least one keyword is not in template_fields.
276
275
"""
277
276
from airflow.models import Variable
278
277
from operators.fake import FakeOperator
@@ -291,10 +290,10 @@ def foo():
291
290
FakeOperator(task_id="fake", foo=var)
292
291
FakeOperator(task_id="fake2", fizz=var)
293
292
""" ,
293
+ id = "UseJinjaVariableGet | SKIP | variable assignment for multiple calls where at least one keyword is not in template_fields" ,
294
294
),
295
- (
295
+ pytest . param (
296
296
UseJinjaVariableGet ,
297
- # Test that variable assignment works for multiple Calls.
298
297
"""
299
298
from airflow.models import Variable
300
299
from operators.fake import FakeOperator
@@ -313,10 +312,10 @@ def foo():
313
312
FakeOperator(task_id="fake", foo=var)
314
313
FakeOperator(task_id="fake2", foo=var)
315
314
""" ,
315
+ id = "UseJinjaVariableGet | SUCCESS | variable assignment for multiple calls" ,
316
316
),
317
- (
317
+ pytest . param (
318
318
UseJinjaVariableGet ,
319
- # Test that nothing happens if the type of Variable.get Calls parent is not implemented e.g. function call
320
319
"""
321
320
from airflow.models import Variable
322
321
from operators.fake import FakeOperator
@@ -329,10 +328,10 @@ def foo():
329
328
330
329
FakeOperator(task_id="fake", fizz=str(Variable.get("FOO")))
331
330
""" ,
331
+ id = "UseJinjaVariableGet | SKIP | direct assignment with unimplemented parent type" ,
332
332
),
333
- (
333
+ pytest . param (
334
334
UseJinjaVariableGet ,
335
- # Test that Variable.get calls with deserialize_json works.
336
335
"""
337
336
from airflow.models import Variable
338
337
from operators.fake import FakeOperator
@@ -345,10 +344,10 @@ def foo():
345
344
346
345
FakeOperator(task_id="fake", foo='{{ var.json.FOO }}')
347
346
""" ,
347
+ id = "UseJinjaVariableGet | SUCCESS | direct assignment with key and deserialize_json keyword" ,
348
348
),
349
- (
349
+ pytest . param (
350
350
UseJinjaVariableGet ,
351
- # Test that Variable.get calls with default_var works.
352
351
"""
353
352
from airflow.models import Variable
354
353
from operators.fake import FakeOperator
@@ -361,10 +360,10 @@ def foo():
361
360
362
361
FakeOperator(task_id="fake", foo="{{ var.value.get('FOO', 'BAR') }}")
363
362
""" ,
363
+ id = "UseJinjaVariableGet | SUCCESS | direct assignment with key and default_var keyword" ,
364
364
),
365
- (
365
+ pytest . param (
366
366
UseJinjaVariableGet ,
367
- # Test that Variable.get calls with default_var=None works.
368
367
"""
369
368
from airflow.models import Variable
370
369
from operators.fake import FakeOperator
@@ -377,10 +376,10 @@ def foo():
377
376
378
377
FakeOperator(task_id="fake", foo="{{ var.value.get('FOO', None) }}")
379
378
""" ,
379
+ id = "UseJinjaVariableGet | SUCCESS | direct assignment with key and default_var=None keyword" ,
380
380
),
381
- (
381
+ pytest . param (
382
382
UseJinjaVariableGet ,
383
- # Test that Variable.get calls works with both - deserialize_json and default_var.
384
383
"""
385
384
from airflow.models import Variable
386
385
from operators.fake import FakeOperator
@@ -393,6 +392,7 @@ def foo():
393
392
394
393
FakeOperator(task_id="fake", foo="{{ var.json.get('FOO', 'BAR') }}")
395
394
""" ,
395
+ id = "UseJinjaVariableGet | SUCCESS | direct assignment with key, deserialize_json and default_var keywords" ,
396
396
),
397
397
],
398
398
)
0 commit comments