ashb commented on a change in pull request #22683:
URL: https://github.com/apache/airflow/pull/22683#discussion_r840431972
##########
File path: tests/models/test_baseoperator.py
##########
@@ -877,3 +877,30 @@ def
test_expand_mapped_task_instance_skipped_on_zero(dag_maker, session):
)
assert indices == [(-1, TaskInstanceState.SKIPPED)]
+
+
+def test_mapped_task_applies_default_args_classic(dag_maker):
+ with dag_maker(default_args={"sla": timedelta(minutes=30)}) as dag:
+ MockOperator(task_id="simple", arg1=None, arg2=0)
+ MockOperator.partial(task_id="mapped").expand(arg1=[1], arg2=[2, 3])
+
+ assert dag.get_task("simple").sla == timedelta(minutes=30)
+ assert dag.get_task("mapped").sla == timedelta(minutes=30)
Review comment:
We are about to make SLA not work with mapped operators so
```suggestion
with dag_maker(default_args={"execution_timeout":
timedelta(minutes=30)}) as dag:
MockOperator(task_id="simple", arg1=None, arg2=0)
MockOperator.partial(task_id="mapped").expand(arg1=[1], arg2=[2, 3])
assert dag.get_task("simple").execution_timeout == timedelta(minutes=30)
assert dag.get_task("mapped").execution_timeout == timedelta(minutes=30)
```
##########
File path: tests/models/test_baseoperator.py
##########
@@ -877,3 +877,30 @@ def
test_expand_mapped_task_instance_skipped_on_zero(dag_maker, session):
)
assert indices == [(-1, TaskInstanceState.SKIPPED)]
+
+
+def test_mapped_task_applies_default_args_classic(dag_maker):
+ with dag_maker(default_args={"sla": timedelta(minutes=30)}) as dag:
+ MockOperator(task_id="simple", arg1=None, arg2=0)
+ MockOperator.partial(task_id="mapped").expand(arg1=[1], arg2=[2, 3])
+
+ assert dag.get_task("simple").sla == timedelta(minutes=30)
+ assert dag.get_task("mapped").sla == timedelta(minutes=30)
+
+
+def test_mapped_task_applies_default_args_taskflow(dag_maker):
+ with dag_maker(default_args={"sla": timedelta(minutes=30)}) as dag:
+
+ @dag.task
+ def simple(arg):
+ pass
+
+ @dag.task
+ def mapped(arg):
+ pass
+
+ simple(arg=0)
+ mapped.expand(arg=[1, 2])
+
+ assert dag.get_task("simple").sla == timedelta(minutes=30)
+ assert dag.get_task("mapped").sla == timedelta(minutes=30)
Review comment:
```suggestion
assert dag.get_task("simple").execution_timeout == timedelta(minutes=30)
assert dag.get_task("mapped").execution_timeout == timedelta(minutes=30)
```
##########
File path: tests/models/test_baseoperator.py
##########
@@ -877,3 +877,30 @@ def
test_expand_mapped_task_instance_skipped_on_zero(dag_maker, session):
)
assert indices == [(-1, TaskInstanceState.SKIPPED)]
+
+
+def test_mapped_task_applies_default_args_classic(dag_maker):
+ with dag_maker(default_args={"sla": timedelta(minutes=30)}) as dag:
+ MockOperator(task_id="simple", arg1=None, arg2=0)
+ MockOperator.partial(task_id="mapped").expand(arg1=[1], arg2=[2, 3])
+
+ assert dag.get_task("simple").sla == timedelta(minutes=30)
+ assert dag.get_task("mapped").sla == timedelta(minutes=30)
+
+
+def test_mapped_task_applies_default_args_taskflow(dag_maker):
+ with dag_maker(default_args={"sla": timedelta(minutes=30)}) as dag:
Review comment:
```suggestion
with dag_maker(default_args={"execution_timeout":
timedelta(minutes=30)}) as dag:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
